hdf5-pure 0.32.0

//! Whole-file repack (issue #21): copy an existing HDF5 file into a fresh,
//! compact one, optionally dropping objects.
//!
//! [`File::open_rw`](crate::File::open_rw) deletes objects in place but reclaims
//! space only within a session and cannot return a single deleted-and-closed
//! file's bytes to the OS. Repack is the complementary answer — the same one the
//! HDF5 C ecosystem ships as `h5repack`: it reads every surviving object and
//! rewrites the whole file from scratch through [`FileBuilder`], so the result
//! has no dead space and is strictly smaller when objects are dropped.
//!
//! # Fidelity contract
//!
//! Repack never silently degrades data. Every surviving object is reproduced
//! faithfully — datatype, shape, max-shape, chunking, filters, and byte-exact
//! element data — or the whole operation fails with [`Error::RepackUnsupported`]
//! naming the object and the reason. It refuses rather than approximate.
//! Currently reproducible:
//!
//! - Datasets with fixed-point, floating-point, time, fixed-length string,
//!   bit-field, opaque, compound, enumeration, and array datatypes,
//!   contiguous/compact or chunked.
//! - **Chunked** datasets copy their compressed chunks **verbatim** (chunk by
//!   chunk, never decoded), so *every* filter is preserved byte-exact: deflate,
//!   shuffle, fletcher32, integer **and** float scale-offset, ZFP, SZIP, and
//!   even filters this crate cannot itself apply. The destination always uses a
//!   v4 chunk index (single-chunk / fixed-array / extensible-array) regardless
//!   of the source index type.
//! - **Variable-length** datasets (1D and ND, contiguous/compact as well as
//!   chunked, filtered, or resizable): string-shaped (`is_string: true` and the
//!   MATLAB VLEN-of-1-byte-ASCII-string shape) and non-string sequences over any
//!   base type that embeds no addresses. Each element's exact heap bytes are read
//!   and re-staged through a fresh global heap, preserving charset, padding, the
//!   null-vs-empty distinction, embedded NULs, and non-UTF-8 payloads, and the
//!   source's chunk geometry, filters, and resizability are carried onto the
//!   rebuilt dataset.
//! - Datatypes that *contain* an address without being one — a compound with a
//!   variable-length member, an object-reference member, or both; an array of
//!   such compounds; and nesting of either. The embedded heap references are
//!   re-staged and the embedded object addresses resolved exactly as their
//!   top-level counterparts are, in a single pass, while every other byte of the
//!   element is carried through untouched.
//! - Contiguous/compact **object-reference** datasets, and contiguous/compact
//!   datatypes containing an object-reference member: each stored address is
//!   rewritten to its target object's new location in the compacted file (null
//!   and undefined references are carried verbatim).
//! - Group hierarchy of arbitrary depth.
//! - Attributes representable as [`AttrValue`] (numbers, fixed and
//!   variable-length strings and their arrays), on datasets, groups, and root.
//! - The source file's file-space management strategy (with its page size and
//!   threshold), carried into the compact output as non-persistent — a repacked
//!   file has no free space to persist.
//!
//! The verbatim chunk copy never decodes, so it eliminates the
//! decompress→recompress round-trip and the per-dataset decompression blowup,
//! and a lossy filter survives byte-exact. Two paths still re-encode and so
//! require **lossless** filters: a *contiguous/compact* filtered dataset, and a
//! *sparse* chunked dataset (one with unallocated chunk-grid holes, which the
//! dense verbatim path cannot lay out). A lossy pipeline on either of those is
//! refused.
//!
//! Refused (named, never dropped silently): chunked, filtered, or resizable
//! datasets whose datatype is or contains an object reference (their element
//! addresses are resolved as elements are re-staged, which a compressed chunk
//! would need rewritten in place);
//! region references and
//! non-8-byte object references; an object reference to a dropped object or to a
//! target outside the hard-link hierarchy (a dangling, named-datatype, or region
//! target), and object references in a userblock file (non-zero base address); a
//! non-string vlen sequence whose base type embeds an address (nested vlen or
//! reference); virtual and external data layouts; a lossy filter on the
//! contiguous re-encode or sparse-chunked fallback path; and any attribute whose
//! datatype the reader cannot decode into an [`AttrValue`] (e.g. an enumeration,
//! compound, reference, or boolean attribute). An object that cannot be
//! reproduced fails the repack by name rather than being silently dropped.
//!
//! # Memory
//!
//! Repack is **out-of-core** (issue [#82]): it opens the source with
//! [`File::open_streaming`], reading metadata and one working chunk on demand
//! rather than buffering the whole file, copies each chunked dataset's
//! compressed chunks verbatim one at a time, and streams the output straight to
//! the destination. Peak memory is therefore bounded by a single chunk plus the
//! file's metadata, independent of dataset (or file) size, so a file whose data
//! exceeds available RAM repacks successfully.
//!
//! [#82]: https://github.com/stephenberry/hdf5-pure/issues/82

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::path::Path;
use std::sync::Arc;

use crate::chunked_read::ChunkInfo;
use crate::chunked_write::{ChunkMeta, ChunkProvider};
use crate::convert::TryToUsize;
use crate::data_layout::DataLayout;
use crate::datatype::{Datatype, ReferenceType};
use crate::error::{Error, FormatError};
use crate::filter_pipeline::{
    FILTER_DEFLATE, FILTER_FLETCHER32, FILTER_LZF, FILTER_SCALEOFFSET, FILTER_SHUFFLE,
    FilterPipeline,
};
use crate::reader::{Dataset, File, Group};
use crate::scaleoffset::{self, ScaleOffset};
use crate::source::Source;
use crate::type_builders::{
    AttrValue, DatasetBuilder, FinishedGroup, GroupBuilder, ObjectRefPatch, ObjectRefTarget,
    VlStringElement,
};
use crate::vl_data::{
    EmbeddedVlSlot, VlByteObject, VlenStringReadOptions, embedded_vlen_slots,
    is_vlen_string_datatype,
};
use crate::writer::FileBuilder;

/// Options controlling a [`repack`].
///
/// Built with [`new`](Self::new) and [`drop_path`](Self::drop_path); the fields
/// are private so a future option is an additive change.
#[derive(Debug, Default, Clone)]
pub struct RepackOptions {
    /// Full paths of objects to omit from the output. See
    /// [`drop_path`](Self::drop_path).
    drop: Vec<String>,
}

impl RepackOptions {
    /// Options that drop nothing — a pure compaction copy.
    pub fn new() -> Self {
        Self::default()
    }

    /// Omit the object at `path` from the output (e.g. `"grp/old"` or
    /// `"/grp/old"`; leading and trailing slashes are ignored). Dropping a group
    /// drops its whole subtree. Every listed path must exist in the source, or
    /// the repack fails — a no-op drop is treated as a mistake rather than
    /// silently ignored. Chainable.
    pub fn drop_path(mut self, path: &str) -> Self {
        self.drop.push(path.to_string());
        self
    }

    /// The paths this repack will omit, in the order they were added.
    pub fn drop_paths(&self) -> &[String] {
        &self.drop
    }
}

/// Repack `src` into a new file at `dst`, applying `options`.
///
/// Reads every object of `src` not excluded by [`RepackOptions::drop_path`] and
/// writes them into a fresh, compact file at `dst`. On success `dst` is a normal
/// HDF5 file holding exactly the surviving objects with no dead space.
///
/// The fidelity checks run first: every object is validated while the output is
/// staged, so an [`Error::RepackUnsupported`] (an object that cannot be
/// reproduced faithfully, or a drop path that does not exist) is reported before
/// any byte is written to `dst`. Dataset *chunk bytes*, by contrast, are streamed
/// from `src` to `dst` during the write rather than buffered, so an I/O error
/// reading the source or writing the destination partway through can leave a
/// partial `dst` (remove it and retry).
///
/// See [`Error::RepackUnsupported`] for the objects that cannot be reproduced
/// faithfully.
pub fn repack<P: AsRef<Path>, Q: AsRef<Path>>(
    src: P,
    dst: Q,
    options: &RepackOptions,
) -> Result<(), Error> {
    // Open the source for on-demand streaming reads: metadata and one working
    // chunk are resident at a time, never the whole file. Shared so each streamed
    // dataset's chunk provider can pull from the same handle during the write
    // without an extra open.
    let file = Arc::new(File::open_streaming(src)?);

    // Normalize the drop set to canonical slash-free paths and remember which
    // ones actually match, so an unmatched drop can be reported as an error.
    let drop: BTreeSet<String> = options.drop.iter().map(|p| normalize(p)).collect();
    let mut matched: BTreeSet<String> = BTreeSet::new();

    let mut builder = FileBuilder::new();
    // Carry the source's file-space strategy forward. The repacked file is
    // compact with no free space, so the strategy and its page size/threshold
    // are preserved but `persist` is reset to false — there is nothing to
    // persist, and writing persistent free-space blocks is a separate feature.
    if let Some(info) = file.file_space_info() {
        builder
            .with_file_space_strategy(info.strategy, false, info.threshold)
            .with_file_space_page_size(info.page_size);
    }
    // Map every source object's (relative) header address to its path, so an
    // object-reference dataset can be rewritten to point at the same objects in
    // the compacted output rather than at their stale source addresses.
    let addr_map = build_object_address_map(&file)?;

    let root = file.root();
    populate(
        &mut builder,
        &root,
        "",
        &drop,
        &mut matched,
        &file,
        &addr_map,
    )?;

    // Every requested drop must have named a real object.
    if let Some(missing) = drop.iter().find(|d| !matched.contains(*d)) {
        return Err(Error::RepackUnsupported(format!(
            "drop path does not exist in the source: {missing}"
        )));
    }

    builder.write(dst)?;
    Ok(())
}

/// A destination that group contents can be added to. Implemented for both the
/// top-level [`FileBuilder`] (the root group) and [`GroupBuilder`] (subgroups)
/// so one recursive walk handles every level.
trait GroupSink {
    fn sink_dataset(&mut self, name: &str) -> &mut DatasetBuilder;
    fn sink_add_group(&mut self, group: FinishedGroup);
    fn sink_set_attr(&mut self, name: &str, value: AttrValue);
}

impl GroupSink for FileBuilder {
    fn sink_dataset(&mut self, name: &str) -> &mut DatasetBuilder {
        self.create_dataset(name)
    }
    fn sink_add_group(&mut self, group: FinishedGroup) {
        self.add_group(group);
    }
    fn sink_set_attr(&mut self, name: &str, value: AttrValue) {
        self.set_attr(name, value);
    }
}

impl GroupSink for GroupBuilder {
    fn sink_dataset(&mut self, name: &str) -> &mut DatasetBuilder {
        self.create_dataset(name)
    }
    fn sink_add_group(&mut self, group: FinishedGroup) {
        self.add_group(group);
    }
    fn sink_set_attr(&mut self, name: &str, value: AttrValue) {
        self.set_attr(name, value);
    }
}

/// Copy `src`'s attributes, datasets, and subgroups (recursively) into `sink`,
/// skipping anything whose path is in `drop`. `path` is the slash-free path of
/// `src` itself (empty for the root).
fn populate<S: GroupSink>(
    sink: &mut S,
    src: &Group,
    path: &str,
    drop: &BTreeSet<String>,
    matched: &mut BTreeSet<String>,
    file: &Arc<File>,
    addr_map: &HashMap<u64, String>,
) -> Result<(), Error> {
    // Attributes, in name order for a deterministic output. Refuse if any
    // attribute on this group cannot be represented (and would be dropped).
    let attrs = src.attrs()?;
    let owner = if path.is_empty() {
        "root group".to_string()
    } else {
        format!("group {path}")
    };
    check_attr_completeness(&attrs, &src.attr_names()?, &owner)?;
    for (name, value) in sorted(attrs) {
        sink.sink_set_attr(&name, value);
    }

    // Datasets, sorted by name.
    let mut dataset_names = src.datasets()?;
    dataset_names.sort();
    for name in dataset_names {
        let child_path = join(path, &name);
        if drop.contains(&child_path) {
            matched.insert(child_path);
            continue;
        }
        let ds = src.dataset(&name)?;
        emit_dataset(
            sink.sink_dataset(&name),
            &ds,
            &child_path,
            file,
            drop,
            addr_map,
        )?;
    }

    // Subgroups, sorted by name; built depth-first into a FinishedGroup.
    let mut group_names = src.groups()?;
    group_names.sort();
    for name in group_names {
        let child_path = join(path, &name);
        if drop.contains(&child_path) {
            matched.insert(child_path);
            continue;
        }
        let child = src.group(&name)?;
        let mut gb = GroupBuilder::new(&name);
        populate(&mut gb, &child, &child_path, drop, matched, file, addr_map)?;
        sink.sink_add_group(gb.finish());
    }
    Ok(())
}

/// Capture one dataset's full description and stage it on `db`, or fail with a
/// named [`Error::RepackUnsupported`] if any part cannot be reproduced.
fn emit_dataset(
    db: &mut DatasetBuilder,
    ds: &Dataset,
    path: &str,
    file: &Arc<File>,
    drop: &BTreeSet<String>,
    addr_map: &HashMap<u64, String>,
) -> Result<(), Error> {
    let datatype = ds.datatype()?;
    let dataspace = ds.dataspace()?;
    let layout = ds.data_layout()?;
    let pipeline = ds.filter_pipeline_parsed();

    check_datatype(&datatype, path)?;
    check_layout(&layout, path)?;

    let dims = dataspace.dimensions.clone();
    let n_elements: u64 = dims.iter().product();

    // Every variable-length reference the datatype reaches, whether it *is*
    // variable-length or merely contains one through a compound member or array
    // entry. Both kinds are re-staged rather than copied, so both disqualify the
    // verbatim and fill-value paths below.
    let vlen_slots = embedded_vlen_slots(&datatype).ok_or_else(|| {
        Error::RepackUnsupported(format!(
            "dataset {path}: datatype declares variable-length members its own element \
             size cannot hold"
        ))
    })?;
    // Likewise every object-header address it reaches. Both kinds of address are
    // rewritten rather than copied, so both disqualify the paths that move
    // element bytes unchanged.
    let reference_slots = embedded_reference_slots(&datatype).ok_or_else(|| {
        Error::RepackUnsupported(format!(
            "dataset {path}: datatype declares object references its own element size cannot hold"
        ))
    })?;

    // A user-defined fill value is fixed element bytes in the datatype, so it
    // reproduces exactly on the fixed-size paths below (dense-chunked verbatim,
    // and contiguous/compact/sparse re-encode). The variable-length and
    // object-reference paths that follow rebuild global-heap references and
    // object addresses from scratch, so a fill value there could carry stale heap
    // or address bytes; refuse rather than copy it, matching the crate's
    // never-degrade-on-rewrite contract.
    let fill = ds.defined_fill_bytes()?;
    if fill.is_some() && !(vlen_slots.is_empty() && reference_slots.is_empty()) {
        return Err(Error::RepackUnsupported(format!(
            "dataset {path}: a fill value on a variable-length or object-reference \
             dataset cannot be repacked faithfully"
        )));
    }

    // Variable-length string datasets take a dedicated path: their element
    // references point into the global heap, so they are re-emitted by reading
    // each element's exact heap bytes and re-staging them, not by copying raw
    // element bytes (whose stored heap addresses would go stale on rewrite).
    if is_vlen_string_datatype(&datatype) {
        emit_vlen_string_dataset(db, ds, path, &datatype, &dims, &layout, &pipeline)?;
        // VL-string datasets carry attributes the same way as any other.
        let attrs = ds.attrs()?;
        check_attr_completeness(&attrs, &ds.attr_names()?, &format!("dataset {path}"))?;
        for (name, value) in sorted(attrs) {
            db.set_attr(&name, value);
        }
        return Ok(());
    }

    // Non-string variable-length (sequence) datasets take the same global-heap
    // re-staging path as VL strings: each element's exact heap bytes are read and
    // re-emitted through a fresh global heap, so the stored heap addresses are
    // rebuilt rather than copied stale. Routed here before the verbatim chunk-copy
    // path so a chunked one is refused (not copied with stale references).
    if is_nonstring_vlen(&datatype) {
        emit_vlen_sequence_dataset(db, ds, path, &datatype, &dims, &layout, &pipeline)?;
        let attrs = ds.attrs()?;
        check_attr_completeness(&attrs, &ds.attr_names()?, &format!("dataset {path}"))?;
        for (name, value) in sorted(attrs) {
            db.set_attr(&name, value);
        }
        return Ok(());
    }

    // Object-reference datasets store absolute object-header addresses that would
    // go stale on rewrite, so each reference is resolved to its target's *new*
    // address (via the source address->path map and the writer's path resolution)
    // rather than copied. Routed here before the verbatim chunk-copy path so a
    // chunked one is refused (not copied with stale addresses).
    if is_object_reference(&datatype) {
        emit_object_reference_dataset(db, ds, path, &dims, &layout, file, drop, addr_map)?;
        let attrs = ds.attrs()?;
        check_attr_completeness(&attrs, &ds.attr_names()?, &format!("dataset {path}"))?;
        for (name, value) in sorted(attrs) {
            db.set_attr(&name, value);
        }
        return Ok(());
    }

    // A datatype that merely *contains* a variable-length member or an object
    // reference — a compound with either, or an array of them — embeds the
    // address inside each element. Those addresses go stale on rewrite exactly as
    // a top-level one does, so this dataset is re-staged rather than copied.
    // Routed before the verbatim chunk-copy path so a chunked one is rebuilt, not
    // copied with source addresses (issue #201).
    //
    // Both kinds are handled together rather than in sequence: a compound can
    // carry one of each, and rewriting only the first kind found would leave the
    // other pointing into the source file — reintroducing the very bug this path
    // exists to fix, and skipping the other kind's refusals with it.
    if !vlen_slots.is_empty() || !reference_slots.is_empty() {
        emit_embedded_address_dataset(
            db,
            ds,
            path,
            &datatype,
            &dims,
            &layout,
            &pipeline,
            &vlen_slots,
            &reference_slots,
            file,
            drop,
            addr_map,
        )?;
        let attrs = ds.attrs()?;
        check_attr_completeness(&attrs, &ds.attr_names()?, &format!("dataset {path}"))?;
        for (name, value) in sorted(attrs) {
            db.set_attr(&name, value);
        }
        return Ok(());
    }

    // Past the reference-bearing paths: carry the fill value through the
    // fixed-size storage paths below, both of which reproduce it exactly.
    db.fill = fill;

    // A chunked dataset with allocated chunks is copied chunk-by-chunk, verbatim:
    // each compressed chunk is laid into the output without decoding, so any
    // filter — including lossy ones (float scale-offset, ZFP) and ones this crate
    // cannot itself apply (SZIP, unknown) — is reproduced byte-exact. This avoids
    // the decompress→recompress round-trip and the whole-dataset decompression
    // blowup of the read-raw path. `check_pipeline` is intentionally skipped here:
    // never decoding makes every filter safe to carry. The datatype check above
    // still refuses time/variable-length/reference types, whose reproduction or
    // embedded addresses are unsafe even when copied verbatim.
    if let DataLayout::Chunked {
        chunk_dimensions, ..
    } = &layout
        && n_elements > 0
    {
        let rank = dims.len();
        let chunk_dims: Vec<u64> = chunk_dimensions
            .iter()
            .take(rank)
            .map(|&c| c as u64)
            .collect();

        if let Some(DenseChunkPlan { meta, grid_order }) =
            try_plan_dense_chunks(ds, &dims, &chunk_dims)?
        {
            let maxshape = dataspace
                .max_dimensions
                .as_ref()
                .filter(|ms| *ms != &dims)
                .map(|ms| ms.as_slice());
            let elem_size = datatype.type_size() as usize;
            // Stream the chunks from the source at write time rather than reading
            // them all now: the provider holds an `Arc<File>` and fetches one
            // chunk at a time, so a huge dataset never sits in memory.
            let provider = DatasetChunkProvider {
                file: Arc::clone(file),
                grid_order,
            };
            db.with_raw_chunks_lazy(
                datatype,
                &dims,
                maxshape,
                &chunk_dims,
                elem_size,
                ds.filter_pipeline_message_bytes(),
                meta,
                Box::new(provider),
            );

            // Carry the dataset's attributes, refusing any that cannot be
            // represented.
            let attrs = ds.attrs()?;
            check_attr_completeness(&attrs, &ds.attr_names()?, &format!("dataset {path}"))?;
            for (name, value) in sorted(attrs) {
                db.set_attr(&name, value);
            }
            return Ok(());
        }

        // Sparse (holes) chunked dataset: the verbatim path needs a dense grid,
        // so fall through to the read-raw + re-encode path below. That path
        // re-encodes, so it is only faithful for lossless filters; a lossy
        // pipeline on a sparse dataset is refused by `check_pipeline`.
    }

    // Contiguous/compact, or a sparse chunked dataset: read the decompressed
    // bytes and re-encode. This path can only reproduce lossless filters, so
    // refuse a lossy pipeline before reading.
    check_pipeline(pipeline.as_ref(), path)?;

    if n_elements == 0 {
        // An empty dataset owns no element bytes: carry just the datatype and
        // shape so the reconstructed dataset has the same signature.
        db.with_dtype(datatype).with_shape(&dims);
    } else {
        let raw = ds.read_raw()?;
        db.with_raw_data(datatype, raw, n_elements)
            .with_shape(&dims);
    }

    carry_shape_and_pipeline(
        db,
        &dims,
        dataspace.max_dimensions.as_deref(),
        &layout,
        &pipeline,
    );

    // Carry the dataset's attributes, refusing if any cannot be represented.
    let attrs = ds.attrs()?;
    check_attr_completeness(&attrs, &ds.attr_names()?, &format!("dataset {path}"))?;
    for (name, value) in sorted(attrs) {
        db.set_attr(&name, value);
    }

    Ok(())
}

/// Carry the source dataset's resizability, chunk geometry, and filter pipeline
/// onto the rebuilt dataset, so a repack reproduces the layout it read rather
/// than flattening it. Shared by the fixed-size re-encode path and the
/// variable-length re-staging paths, which both rebuild their elements from
/// scratch and so must reapply the layout themselves.
///
/// `check_pipeline` has already rejected any filter not handled here, so the
/// match over filter ids is exhaustive.
fn carry_shape_and_pipeline(
    db: &mut DatasetBuilder,
    dims: &[u64],
    max_dimensions: Option<&[u64]>,
    layout: &DataLayout,
    pipeline: &Option<FilterPipeline>,
) {
    // A max-shape that differs from the current shape means a resizable dataset.
    if let Some(maxshape) = max_dimensions
        && maxshape != dims
    {
        db.with_maxshape(maxshape);
    }

    // Chunking: the v3 layout appends the element size as a trailing chunk
    // dimension, so keep only the first `rank` entries; v4 already stores `rank`.
    if let DataLayout::Chunked {
        chunk_dimensions, ..
    } = layout
    {
        let rank = dims.len();
        let logical: Vec<u64> = chunk_dimensions
            .iter()
            .take(rank)
            .map(|&c| c as u64)
            .collect();
        db.with_chunks(&logical);
    }

    // Re-apply supported filters in their stored order.
    if let Some(p) = pipeline {
        for f in &p.filters {
            match f.filter_id {
                FILTER_SHUFFLE => {
                    db.with_shuffle();
                }
                FILTER_FLETCHER32 => {
                    db.with_fletcher32();
                }
                FILTER_DEFLATE => {
                    // Client-data[0] is the deflate level; default to 6 if absent.
                    db.with_deflate(f.client_data.first().copied().unwrap_or(6));
                }
                FILTER_LZF => {
                    db.with_lzf();
                }
                FILTER_SCALEOFFSET => {
                    // `check_pipeline` guarantees integer (lossless) mode here.
                    // Re-apply with the source's minbits parameter; integer
                    // scale-offset reconstructs the exact element bytes.
                    if let Some(mode @ ScaleOffset::Integer(_)) =
                        scaleoffset::scale_offset_mode(&f.client_data)
                    {
                        db.with_scale_offset(mode);
                    } else {
                        unreachable!("check_pipeline rejected non-integer scale-offset");
                    }
                }
                _ => unreachable!("check_pipeline rejected unsupported filters"),
            }
        }
    }
}

/// Re-emit a variable-length string dataset faithfully: read each element's
/// exact heap bytes (preserving null-vs-empty, charset, padding, and the source
/// VL datatype shape) and re-stage them through the writer's VL-string path,
/// then reapply the source's chunk geometry, filters, and resizability.
///
/// The rebuilt references carry the *new* file's heap addresses. For a chunked
/// or filtered layout the writer places those collections ahead of the chunk
/// data so the addresses are known before encoding (issue #109), which is what
/// makes this path reproduce such a dataset rather than refuse it.
fn emit_vlen_string_dataset(
    db: &mut DatasetBuilder,
    ds: &Dataset,
    path: &str,
    datatype: &Datatype,
    dims: &[u64],
    layout: &DataLayout,
    pipeline: &Option<FilterPipeline>,
) -> Result<(), Error> {
    // A lossy pipeline cannot be reproduced, so refuse it before reading — the
    // same guard the fixed-size re-encode path applies.
    check_pipeline(pipeline.as_ref(), path)?;

    // Read each element's exact heap bytes, preserving the null-vs-empty
    // distinction. Reading bytes (not the lossily UTF-8-decoded `String`) keeps
    // embedded NULs and non-UTF-8 payloads byte-exact.
    let objects = ds.read_vlen_string_bytes(VlenStringReadOptions::default())?;
    let elements: Vec<VlStringElement> = objects
        .into_iter()
        .map(|o| match o {
            VlByteObject::Null => VlStringElement::Null,
            VlByteObject::Bytes(bytes) => VlStringElement::Bytes(bytes),
        })
        .collect();

    // Re-stage with the exact source datatype, then set the shape. ND datasets
    // round-trip because the element references are stored row-major, matching
    // the order `read_vlen_string_bytes` returns.
    db.with_vlen_string_elements(datatype.clone(), &elements)
        .map_err(Error::Format)?;
    db.with_shape(dims);
    carry_shape_and_pipeline(
        db,
        dims,
        ds.dataspace()?.max_dimensions.as_deref(),
        layout,
        pipeline,
    );
    Ok(())
}

/// Re-emit a non-string variable-length (sequence) dataset faithfully: read each
/// element's exact heap bytes and re-stage them through a fresh global heap, so
/// the rewritten file's heap addresses are rebuilt rather than copied stale.
///
/// A chunked, filtered, or resizable layout is reproduced rather than refused:
/// sequences stage through the same global-heap path as VL strings, so they
/// inherit the early collection placement that makes the heap addresses known
/// before the chunks are encoded (issue #109).
fn emit_vlen_sequence_dataset(
    db: &mut DatasetBuilder,
    ds: &Dataset,
    path: &str,
    datatype: &Datatype,
    dims: &[u64],
    layout: &DataLayout,
    pipeline: &Option<FilterPipeline>,
) -> Result<(), Error> {
    check_pipeline(pipeline.as_ref(), path)?;

    // Read each element's exact heap bytes (preserving the null-vs-empty
    // distinction and any embedded NULs), then re-stage with the source datatype.
    let (objects, _element_size) = ds.read_vlen_sequence_bytes(VlenStringReadOptions::default())?;
    let elements: Vec<VlStringElement> = objects
        .into_iter()
        .map(|o| match o {
            VlByteObject::Null => VlStringElement::Null,
            VlByteObject::Bytes(bytes) => VlStringElement::Bytes(bytes),
        })
        .collect();

    db.with_vlen_sequence_elements(datatype.clone(), &elements)
        .map_err(Error::Format)?;
    db.with_shape(dims);
    carry_shape_and_pipeline(
        db,
        dims,
        ds.dataspace()?.max_dimensions.as_deref(),
        layout,
        pipeline,
    );
    Ok(())
}

/// Re-emit a dataset whose datatype *contains* an address without being one — a
/// compound with a variable-length or object-reference member, an array of such
/// compounds, or nesting of either (issue #201).
///
/// The element bytes are read as they stand; each embedded variable-length
/// reference is resolved against the source's global heap and its payload
/// re-staged into the destination's own, and each embedded object address is
/// resolved to the path it names so the writer can fill in the target's *new*
/// location. Copying the element bytes verbatim instead would leave every
/// address pointing into the source file, which the destination never
/// reproduces — a file that reads back plausibly here and not at all in the
/// reference C library.
///
/// Both kinds are handled in one pass because a single compound can carry one of
/// each, and they patch disjoint byte ranges of the same buffer. Everything
/// outside those ranges is carried byte-for-byte, so the fixed-size members keep
/// their exact stored bytes and byte order.
#[allow(clippy::too_many_arguments)]
fn emit_embedded_address_dataset(
    db: &mut DatasetBuilder,
    ds: &Dataset,
    path: &str,
    datatype: &Datatype,
    dims: &[u64],
    layout: &DataLayout,
    pipeline: &Option<FilterPipeline>,
    vlen_slots: &[EmbeddedVlSlot],
    reference_slots: &[usize],
    file: &Arc<File>,
    drop: &BTreeSet<String>,
    addr_map: &HashMap<u64, String>,
) -> Result<(), Error> {
    // This path re-encodes, so a lossy filter cannot be reproduced — the same
    // guard the other re-staging paths apply.
    check_pipeline(pipeline.as_ref(), path)?;

    // An embedded object address is resolved as the elements are re-staged, which
    // a compressed chunk would need rewritten in place, so the layout guards that
    // protect a top-level object-reference dataset apply here too. They are
    // checked before anything is read, and before the variable-length work, so a
    // compound carrying both kinds is refused rather than half-rewritten.
    if !reference_slots.is_empty() {
        check_embedded_reference_layout(ds, path, dims, layout, file)?;
    }

    let n_elements: u64 = dims.iter().product();

    // Read the element bytes once, resolving the variable-length payloads in the
    // same pass when there are any.
    let (raw, vl_offsets, vl_elements) = if vlen_slots.is_empty() {
        let raw = if n_elements == 0 {
            Vec::new()
        } else {
            ds.read_raw()?
        };
        (raw, Vec::new(), Vec::new())
    } else {
        let data = ds.read_embedded_vlen_bytes(vlen_slots, VlenStringReadOptions::default())?;
        let elements: Vec<VlStringElement> = data
            .objects
            .into_iter()
            .map(|o| match o {
                VlByteObject::Null => VlStringElement::Null,
                VlByteObject::Bytes(bytes) => VlStringElement::Bytes(bytes),
            })
            .collect();
        (data.raw, data.offsets, elements)
    };

    // Resolve the object addresses from the bytes as read, before the
    // variable-length staging below consumes `raw`. The two kinds occupy disjoint
    // slots, so reading one is unaffected by rewriting the other.
    let reference_patches =
        resolve_embedded_references(&raw, datatype, dims, path, drop, addr_map, reference_slots)?;

    if vlen_slots.is_empty() {
        db.with_embedded_object_references(datatype.clone(), raw, n_elements, reference_patches);
    } else {
        db.with_embedded_vlen_elements(
            datatype.clone(),
            raw,
            n_elements,
            &vl_offsets,
            &vl_elements,
        );
        if !reference_patches.is_empty() {
            db.reference_targets = Some(reference_patches);
        }
    }
    db.with_shape(dims);
    carry_shape_and_pipeline(
        db,
        dims,
        ds.dataspace()?.max_dimensions.as_deref(),
        layout,
        pipeline,
    );
    Ok(())
}

/// Turn every embedded object address in `raw` into the target the writer
/// resolves at serialization time.
fn resolve_embedded_references(
    raw: &[u8],
    datatype: &Datatype,
    dims: &[u64],
    path: &str,
    drop: &BTreeSet<String>,
    addr_map: &HashMap<u64, String>,
    slots: &[usize],
) -> Result<Vec<ObjectRefPatch>, Error> {
    if slots.is_empty() {
        return Ok(Vec::new());
    }
    let stride = datatype.type_size() as usize;
    let n_elements: usize = dims.iter().product::<u64>().to_usize()?;
    let needed = n_elements
        .checked_mul(stride)
        .ok_or(FormatError::OffsetOverflow {
            offset: n_elements as u64,
            length: stride as u64,
        })?;
    if raw.len() < needed {
        return Err(FormatError::UnexpectedEof {
            expected: needed,
            available: raw.len(),
        }
        .into());
    }

    let mut patches = Vec::with_capacity(n_elements * slots.len());
    for e in 0..n_elements {
        for &slot in slots {
            let at = e * stride + slot;
            let v = u64::from_le_bytes(
                raw[at..at + 8]
                    .try_into()
                    .expect("slot offsets leave 8 bytes inside the element"),
            );
            patches.push(ObjectRefPatch {
                byte_offset: at,
                target: resolve_reference_address(v, path, drop, addr_map)?,
            });
        }
    }
    Ok(patches)
}

/// A [`ChunkProvider`] that streams a dense chunked dataset's chunks from the
/// source file one at a time during the write, so repack never holds more than a
/// single chunk's bytes. Holds an `Arc<File>` (so it owns its source with no
/// borrowed lifetime) and the source [`ChunkInfo`] for each grid slot.
struct DatasetChunkProvider {
    file: Arc<File>,
    /// Source chunk descriptors in dense row-major grid order, one per slot.
    grid_order: Vec<ChunkInfo>,
}

impl ChunkProvider for DatasetChunkProvider {
    fn chunk_bytes(&self, index: usize, out: &mut Vec<u8>) -> Result<(), FormatError> {
        // Read exactly the chunk's compressed bytes at its recorded address, with
        // no decode and no `addr_offset` adjustment — the same slice the chunked
        // reader consumes. `read_at` fills the whole buffer or errors, and the
        // emitter additionally checks the length against the planned size, so the
        // layout cannot silently desync from the data. Reading straight into the
        // emitter's reused buffer keeps repack at one chunk-sized allocation for
        // the whole dataset.
        let info = &self.grid_order[index];
        let source = self.file.source();
        let len = info.chunk_size as usize;
        // Bounds-check before growing the buffer, the way `Source::read_exact_at`
        // does and for its reason: `chunk_size` comes from the source's chunk
        // index, so a malformed file could name a 4 GiB chunk and have this zero
        // that much memory only for the read to fail EOF anyway.
        let end = info
            .address
            .checked_add(len as u64)
            .ok_or(FormatError::OffsetOverflow {
                offset: info.address,
                length: len as u64,
            })?;
        if end > source.len() {
            return Err(FormatError::UnexpectedEof {
                expected: end.to_usize().unwrap_or(usize::MAX),
                available: source.len().to_usize().unwrap_or(usize::MAX),
            });
        }
        let start = out.len();
        out.resize(start + len, 0);
        source.read_at(info.address, &mut out[start..])
    }
}

/// A planned dense chunked dataset: per-chunk sizes/masks (enough to lay out the
/// destination) plus the source chunk descriptors, both in dense grid order.
struct DenseChunkPlan {
    meta: Vec<ChunkMeta>,
    grid_order: Vec<ChunkInfo>,
}

/// Plan a chunked dataset's verbatim copy without reading any chunk bytes: if
/// every chunk-grid slot is present exactly once (a dense grid), return the
/// per-chunk [`ChunkMeta`] (sizes + filter masks) and the source [`ChunkInfo`]
/// for each slot, both in dense row-major grid order. Returns `Ok(None)` when
/// the grid has holes (a sparse dataset), so the caller falls back to read-raw.
///
/// `dims` is the dataspace shape; `chunk_dims` the logical (rank-only) chunk
/// dimensions. The grid has `num_chunks_per_dim[d] = ceil(dims[d]/chunk_dims[d])`
/// slots per dimension; a chunk at N-d offset `o` maps to grid coordinate
/// `o[d]/chunk_dims[d]` and linear (row-major) index over the grid.
fn try_plan_dense_chunks(
    ds: &Dataset,
    dims: &[u64],
    chunk_dims: &[u64],
) -> Result<Option<DenseChunkPlan>, Error> {
    // Map the source chunks onto the dense grid via the shared planner (the
    // single owner of grid-mapping logic, also used by the in-place editor); a
    // sparse grid (holes/duplicates/misalignment) returns `None`.
    let Some(grid) = crate::chunked_read::plan_dense_grid(ds.raw_chunks()?, dims, chunk_dims)
    else {
        return Ok(None);
    };
    let grid_order = grid.grid_order;
    let meta = grid_order
        .iter()
        .map(|info| ChunkMeta {
            compressed_size: u64::from(info.chunk_size),
            filter_mask: info.filter_mask,
        })
        .collect();
    Ok(Some(DenseChunkPlan { meta, grid_order }))
}

/// Refuse the repack if `owner` has an attribute the reader cannot represent as
/// an [`AttrValue`] and would therefore drop. `names` is every attribute on the
/// object; `decoded` is the subset that read back, keyed by name. Any name not
/// in `decoded` is an attribute that would be silently lost.
fn check_attr_completeness(
    decoded: &std::collections::HashMap<String, AttrValue>,
    names: &[String],
    owner: &str,
) -> Result<(), Error> {
    for name in names {
        if !decoded.contains_key(name) {
            return Err(Error::RepackUnsupported(format!(
                "{owner}: attribute {name:?} has a datatype that cannot be repacked faithfully yet"
            )));
        }
    }
    Ok(())
}

/// Reject datatypes whose on-disk form this crate cannot re-emit faithfully,
/// recursing into compound members, enumeration bases, and array element types so
/// a nested occurrence is caught too. Region and non-8-byte object references are
/// the remaining refusals (their stored selections/addresses are not yet
/// rewritten); 8-byte object references are handled by the reference rewrite path.
fn check_datatype(dt: &Datatype, path: &str) -> Result<(), Error> {
    let bad = |what: &str| {
        Err(Error::RepackUnsupported(format!(
            "dataset {path}: {what} datatype cannot be repacked faithfully yet"
        )))
    };
    match dt {
        // Scalar and opaque-bytes datatypes whose on-disk form `Datatype::serialize`
        // reproduces exactly (including the time type's byte order), so reading the
        // raw element bytes and re-emitting them is byte-for-byte faithful.
        Datatype::FixedPoint { .. }
        | Datatype::FloatingPoint { .. }
        | Datatype::Time { .. }
        | Datatype::String { .. }
        | Datatype::BitField { .. }
        | Datatype::Opaque { .. } => Ok(()),
        // String-shaped variable-length datatypes (`is_string: true`, or the
        // MATLAB VLEN-of-1-byte-ASCII-string shape) are reproduced by reading
        // each element's exact heap bytes and re-staging them through the
        // writer's VL-string path; the layout/filter checks gate chunked ones.
        Datatype::VariableLength { .. } if is_vlen_string_datatype(dt) => Ok(()),
        // Non-string VL (sequences of arbitrary base types) are re-staged the
        // same way, but only when the base type's bytes carry no embedded heap or
        // file addresses that a verbatim copy would leave stale.
        Datatype::VariableLength { base_type, .. } => check_vlen_base_type(base_type, path),
        // Object references (8-byte object-header addresses) are repacked by
        // rewriting each address to its target's new location. Region references
        // (which embed a dataspace selection in the global heap) and non-8-byte
        // object references are not reproduced yet.
        Datatype::Reference {
            ref_type: ReferenceType::Object,
            size: 8,
        } => Ok(()),
        Datatype::Reference {
            ref_type: ReferenceType::Object,
            ..
        } => bad("non-8-byte object reference"),
        Datatype::Reference {
            ref_type: ReferenceType::DatasetRegion,
            ..
        } => bad("dataset-region reference"),
        Datatype::Compound { members, .. } => {
            for m in members {
                check_datatype(&m.datatype, path)?;
            }
            Ok(())
        }
        Datatype::Enumeration { base_type, .. } => check_datatype(base_type, path),
        Datatype::Array { base_type, .. } => check_datatype(base_type, path),
    }
}

/// Whether `dt` is a non-string variable-length (sequence) datatype — the kind
/// re-emitted by [`emit_vlen_sequence_dataset`]. Excludes the string-shaped VL
/// datatypes, which [`emit_vlen_string_dataset`] handles.
fn is_nonstring_vlen(dt: &Datatype) -> bool {
    matches!(dt, Datatype::VariableLength { .. }) && !is_vlen_string_datatype(dt)
}

/// A non-string VL sequence is repacked by re-staging each element's exact heap
/// bytes verbatim. That is faithful only when the base type's bytes embed no
/// addresses that would go stale on rewrite: a nested variable-length type (its
/// elements are themselves global-heap references) and a reference (a stale file
/// address) are refused, recursing through compound members, array elements, and
/// enumeration bases so a nested occurrence is caught too.
fn check_vlen_base_type(dt: &Datatype, path: &str) -> Result<(), Error> {
    let bad = |what: &str| {
        Err(Error::RepackUnsupported(format!(
            "dataset {path}: variable-length sequence of {what} cannot be repacked faithfully yet"
        )))
    };
    match dt {
        Datatype::FixedPoint { .. }
        | Datatype::FloatingPoint { .. }
        | Datatype::Time { .. }
        | Datatype::String { .. }
        | Datatype::BitField { .. }
        | Datatype::Opaque { .. } => Ok(()),
        Datatype::Reference { .. } => bad("references"),
        Datatype::VariableLength { .. } => bad("variable-length elements"),
        Datatype::Compound { members, .. } => {
            for m in members {
                check_vlen_base_type(&m.datatype, path)?;
            }
            Ok(())
        }
        Datatype::Enumeration { base_type, .. } => check_vlen_base_type(base_type, path),
        Datatype::Array { base_type, .. } => check_vlen_base_type(base_type, path),
    }
}

/// Whether `dt` is an object-reference datatype handled by
/// [`emit_object_reference_dataset`].
fn is_object_reference(dt: &Datatype) -> bool {
    matches!(
        dt,
        Datatype::Reference {
            ref_type: ReferenceType::Object,
            ..
        }
    )
}

/// Whether `path` is dropped from the output: either listed in `drop`, or nested
/// under a dropped group (so its whole subtree is gone).
fn is_dropped(path: &str, drop: &BTreeSet<String>) -> bool {
    if drop.contains(path) {
        return true;
    }
    let mut p = path;
    while let Some(idx) = p.rfind('/') {
        p = &p[..idx];
        if drop.contains(p) {
            return true;
        }
    }
    false
}

/// Build a map from each source object's header address to its slash-free path,
/// for resolving object references. With a zero base address (the case object
/// references are repacked for) the stored reference value is exactly this
/// header address, so the lookup is direct.
fn build_object_address_map(file: &File) -> Result<HashMap<u64, String>, Error> {
    let mut map = HashMap::new();
    let root = file.root();
    // The root group can itself be referenced (the writer registers it under the
    // empty path).
    map.insert(root.header_address(), String::new());
    collect_addresses(&root, "", &mut map)?;
    Ok(map)
}

/// Recursively record `(header address -> path)` for every dataset and subgroup.
fn collect_addresses(
    group: &Group,
    prefix: &str,
    map: &mut HashMap<u64, String>,
) -> Result<(), Error> {
    for name in group.datasets()? {
        let ds = group.dataset(&name)?;
        map.insert(ds.header_address(), join(prefix, &name));
    }
    for name in group.groups()? {
        let child = group.group(&name)?;
        let child_path = join(prefix, &name);
        map.insert(child.header_address(), child_path.clone());
        collect_addresses(&child, &child_path, map)?;
    }
    Ok(())
}

/// Re-emit an object-reference dataset faithfully: rewrite each stored address to
/// point at its target's destination location instead of its stale source one.
///
/// Each reference is read, resolved through `addr_map` to a source path, and
/// re-staged as a path target the writer resolves once destination addresses are
/// known. Null (address 0) and undefined (`HADDR_UNDEF`) references are carried
/// verbatim. Refused by name: chunked/filtered or resizable layouts, a non-zero
/// base address, a reference to a dropped object, and a reference whose target is
/// not a hard-linked group or dataset in the source (dangling, or a named
/// datatype / region target not modelled yet).
#[allow(clippy::too_many_arguments)]
fn emit_object_reference_dataset(
    db: &mut DatasetBuilder,
    ds: &Dataset,
    path: &str,
    dims: &[u64],
    layout: &DataLayout,
    file: &Arc<File>,
    drop: &BTreeSet<String>,
    addr_map: &HashMap<u64, String>,
) -> Result<(), Error> {
    if matches!(layout, DataLayout::Chunked { .. }) {
        return Err(Error::RepackUnsupported(format!(
            "dataset {path}: chunked or filtered object-reference datasets cannot be repacked \
             (their addresses live inside compressed chunks and would need rewriting in place)"
        )));
    }
    if let Some(maxshape) = &ds.dataspace()?.max_dimensions
        && maxshape != dims
    {
        return Err(Error::RepackUnsupported(format!(
            "dataset {path}: resizable object-reference datasets cannot be repacked"
        )));
    }
    // Object references store addresses relative to the base address; the rewrite
    // path assumes a zero base (the universal case), so a userblock file is
    // refused rather than risk a mis-resolved address.
    if file.base_address() != 0 {
        return Err(Error::RepackUnsupported(format!(
            "dataset {path}: object references in a file with a non-zero base address (userblock) \
             cannot be repacked yet"
        )));
    }

    let n_elements: usize = dims.iter().product::<u64>().to_usize()?;
    let targets = if n_elements == 0 {
        Vec::new()
    } else {
        let raw = ds.read_raw()?;
        let needed = n_elements
            .checked_mul(8)
            .ok_or(FormatError::OffsetOverflow {
                offset: n_elements as u64,
                length: 8,
            })?;
        if raw.len() < needed {
            return Err(FormatError::UnexpectedEof {
                expected: needed,
                available: raw.len(),
            }
            .into());
        }
        let mut targets = Vec::with_capacity(n_elements);
        for chunk in raw[..needed].chunks_exact(8) {
            let v = u64::from_le_bytes(chunk.try_into().expect("chunks_exact(8) yields 8 bytes"));
            targets.push(resolve_reference_address(v, path, drop, addr_map)?);
        }
        targets
    };

    db.with_object_references(targets);
    db.with_shape(dims);
    Ok(())
}

/// The layout guards that protect a dataset carrying an embedded object address.
///
/// The address is resolved as the elements are re-staged, which a compressed
/// chunk would need rewritten in place, and a userblock shifts every address by a
/// base this rewrite path does not model. Each is refused by name rather than
/// risking a mis-resolved address.
fn check_embedded_reference_layout(
    ds: &Dataset,
    path: &str,
    dims: &[u64],
    layout: &DataLayout,
    file: &Arc<File>,
) -> Result<(), Error> {
    if matches!(layout, DataLayout::Chunked { .. }) {
        return Err(Error::RepackUnsupported(format!(
            "dataset {path}: chunked or filtered datasets with an object-reference member cannot \
             be repacked (their addresses live inside compressed chunks and would need rewriting \
             in place)"
        )));
    }
    if let Some(maxshape) = &ds.dataspace()?.max_dimensions
        && maxshape != dims
    {
        return Err(Error::RepackUnsupported(format!(
            "dataset {path}: resizable datasets with an object-reference member cannot be repacked"
        )));
    }
    if file.base_address() != 0 {
        return Err(Error::RepackUnsupported(format!(
            "dataset {path}: object references in a file with a non-zero base address (userblock) \
             cannot be repacked yet"
        )));
    }
    Ok(())
}

/// Turn one stored object-reference address into the target the writer resolves
/// at serialization time. Null (0) and undefined (`HADDR_UNDEF`) point at nothing
/// and are carried verbatim; anything else must name a hard-linked object that
/// survives the repack.
fn resolve_reference_address(
    address: u64,
    path: &str,
    drop: &BTreeSet<String>,
    addr_map: &HashMap<u64, String>,
) -> Result<ObjectRefTarget, Error> {
    if address == 0 || address == u64::MAX {
        return Ok(ObjectRefTarget::Raw(address));
    }
    match addr_map.get(&address) {
        Some(target_path) if is_dropped(target_path, drop) => {
            Err(Error::RepackUnsupported(format!(
                "dataset {path}: object reference to dropped object {target_path:?} cannot be repacked"
            )))
        }
        Some(target_path) => Ok(ObjectRefTarget::Path(target_path.clone())),
        None => Err(Error::RepackUnsupported(format!(
            "dataset {path}: object reference to address {address:#x} resolves to no hard-linked \
             object in the source (dangling, or a named-datatype / region target not supported yet)"
        ))),
    }
}

/// Every 8-byte object reference `datatype` reaches through a compound member or
/// array entry, as byte offsets within one element, in declaration order.
///
/// Mirrors [`embedded_vlen_slots`] for the other kind of address a rewrite
/// invalidates. A datatype that *is* an object reference yields the single slot
/// at offset 0, so callers handling that case separately should test for it
/// first. Returns `None` if the offsets found do not fit the datatype's declared
/// element size, which means the element bytes cannot be walked safely.
fn embedded_reference_slots(datatype: &Datatype) -> Option<Vec<usize>> {
    /// Returns `false` when the datatype cannot be walked on this target, for the
    /// reasons [`embedded_vlen_slots`]' walker documents.
    fn collect(datatype: &Datatype, base: usize, capacity: usize, out: &mut Vec<usize>) -> bool {
        if out.len() > capacity {
            return true;
        }
        match datatype {
            Datatype::Reference {
                ref_type: ReferenceType::Object,
                size: 8,
            } => {
                out.push(base);
                true
            }
            Datatype::Compound { members, .. } => {
                for m in members {
                    let Some(at) = usize::try_from(m.byte_offset)
                        .ok()
                        .and_then(|off| base.checked_add(off))
                    else {
                        return false;
                    };
                    if !collect(&m.datatype, at, capacity, out) {
                        return false;
                    }
                }
                true
            }
            Datatype::Array {
                base_type,
                dimensions,
            } => {
                // As in `embedded_vlen_slots`: probe once so that entries which can
                // never contribute do not drive a walk over huge declared
                // dimensions, and so every iteration below pushes at least one slot.
                // Walked once and translated per entry, for the reason
                // `embedded_vlen_slots` documents: re-walking is exponential in
                // nesting depth.
                let mut probe = Vec::new();
                if !collect(base_type, 0, capacity, &mut probe) {
                    return false;
                }
                if probe.is_empty() {
                    return true;
                }
                let count = dimensions
                    .iter()
                    .copied()
                    .fold(1u64, |a, b| a.saturating_mul(u64::from(b)));
                // As in `embedded_vlen_slots`: more entries than the element has
                // room for cannot fit, so reject without walking them.
                if count > capacity as u64 {
                    return false;
                }
                let entries = usize::try_from(count).unwrap_or(usize::MAX);
                let stride = base_type.type_size() as usize;
                for i in 0..entries {
                    let Some(at) = i.checked_mul(stride).and_then(|off| base.checked_add(off))
                    else {
                        return false;
                    };
                    for &slot in &probe {
                        let Some(off) = at.checked_add(slot) else {
                            return false;
                        };
                        out.push(off);
                        if out.len() > capacity {
                            return true;
                        }
                    }
                }
                true
            }
            _ => true,
        }
    }

    let element_size = datatype.type_size() as usize;
    let capacity = element_size / 8;
    let mut slots = Vec::new();
    if !collect(datatype, 0, capacity, &mut slots) {
        return None;
    }
    // `checked_add`: an offset near the top of the address space would otherwise
    // wrap here and read as "fits".
    if slots.len() > capacity
        || slots
            .iter()
            .any(|&s| s.checked_add(8).is_none_or(|end| end > element_size))
    {
        return None;
    }
    Some(slots)
}

/// Reject data layouts that cannot be read and re-emitted (virtual datasets;
/// contiguous/chunked with an undefined address are allowed — they are empty).
fn check_layout(layout: &DataLayout, path: &str) -> Result<(), Error> {
    match layout {
        DataLayout::Compact { .. } | DataLayout::Contiguous { .. } | DataLayout::Chunked { .. } => {
            Ok(())
        }
        DataLayout::Virtual { .. } => Err(Error::RepackUnsupported(format!(
            "dataset {path}: virtual data layout cannot be repacked"
        ))),
    }
}

/// Reject any filter that cannot be reproduced **by the re-encoding path**, so a
/// filtered dataset is never silently rewritten without its filters.
///
/// This guards only the two paths that read each dataset's *decompressed* bytes
/// and re-apply its filters from scratch: a contiguous/compact filtered dataset,
/// and the sparse-chunked fallback. A filter is safe there only when it is
/// **lossless** — then the re-encoded chunks decompress to the exact same bytes.
/// Deflate, shuffle, fletcher32, LZF, and integer scale-offset qualify. Float D-scale
/// scale-offset and ZFP are lossy: re-encoding already-decompressed values is not
/// guaranteed idempotent, so reproducing them could silently perturb the data,
/// and they are refused. SZIP this crate cannot write at all.
///
/// The dense chunked path (the common case) copies compressed chunks verbatim
/// and never calls this — there every filter is safe because nothing is decoded.
fn check_pipeline(pipeline: Option<&FilterPipeline>, path: &str) -> Result<(), Error> {
    let Some(p) = pipeline else {
        return Ok(());
    };
    // Re-encoding replays filters through the builder, which emits its own
    // fixed order and refuses lzf + deflate on one dataset, so a foreign
    // pipeline carrying both cannot be reproduced faithfully.
    let has = |id| p.filters.iter().any(|f| f.filter_id == id);
    if has(FILTER_LZF) && has(FILTER_DEFLATE) {
        return Err(Error::RepackUnsupported(format!(
            "dataset {path}: an lzf + deflate pipeline cannot be re-encoded faithfully"
        )));
    }
    for f in &p.filters {
        match f.filter_id {
            FILTER_DEFLATE | FILTER_SHUFFLE | FILTER_FLETCHER32 | FILTER_LZF => {}
            FILTER_SCALEOFFSET => match scaleoffset::scale_offset_mode(&f.client_data) {
                Some(ScaleOffset::Integer(_)) => {}
                _ => {
                    return Err(Error::RepackUnsupported(format!(
                        "dataset {path}: only lossless integer scale-offset with an undefined fill value can be repacked faithfully"
                    )));
                }
            },
            other => {
                return Err(Error::RepackUnsupported(format!(
                    "dataset {path}: filter id {other} cannot be repacked yet"
                )));
            }
        }
    }
    Ok(())
}

/// Sort a name→value attribute map into a deterministic, ordered list.
fn sorted(attrs: std::collections::HashMap<String, AttrValue>) -> Vec<(String, AttrValue)> {
    attrs
        .into_iter()
        .collect::<BTreeMap<_, _>>()
        .into_iter()
        .collect()
}

/// Canonicalize a path to slash-free form: split on `/`, drop empty components,
/// rejoin. `"/a//b/"` and `"a/b"` both become `"a/b"`.
fn normalize(path: &str) -> String {
    path.split('/')
        .filter(|c| !c.is_empty())
        .collect::<Vec<_>>()
        .join("/")
}

/// Join a parent path (slash-free, possibly empty) with a child name.
fn join(parent: &str, name: &str) -> String {
    if parent.is_empty() {
        name.to_string()
    } else {
        format!("{parent}/{name}")
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A foreign pipeline carrying both lzf and deflate is refused with the
    /// repack error, not the builder's combination error: the builder replays
    /// filters in its own fixed order, so the stored order cannot be
    /// reproduced. The builder cannot produce such a file, hence a synthetic
    /// pipeline rather than a file-level test.
    #[test]
    fn lzf_plus_deflate_pipeline_is_refused() {
        use crate::filter_pipeline::FilterDescription;

        let pipeline = FilterPipeline {
            version: 2,
            filters: vec![
                FilterDescription {
                    filter_id: FILTER_LZF,
                    name: Some("lzf".into()),
                    flags: 0,
                    client_data: vec![],
                },
                FilterDescription {
                    filter_id: FILTER_DEFLATE,
                    name: None,
                    flags: 0,
                    client_data: vec![6],
                },
            ],
        };
        let err = check_pipeline(Some(&pipeline), "d").unwrap_err();
        assert!(
            matches!(&err, Error::RepackUnsupported(msg) if msg.contains("lzf + deflate")),
            "unexpected error: {err:?}"
        );
    }

    #[test]
    fn repack_preserves_big_endian_time_dataset() {
        // The reference C library cannot create H5T_TIME, so this round-trips a
        // big-endian time dataset through our own writer and reader: repack must
        // preserve both the byte order (bf0 bit 0) and the raw element bytes.
        use crate::datatype::{Datatype, DatatypeByteOrder};
        use crate::reader::File;
        use crate::writer::FileBuilder;

        let dir = std::env::temp_dir();
        let src = dir.join("hdf5_pure_repack_time_src.h5");
        let dst = dir.join("hdf5_pure_repack_time_dst.h5");

        let dt = Datatype::Time {
            size: 4,
            byte_order: DatatypeByteOrder::BigEndian,
            bit_precision: 32,
        };
        let raw: Vec<u8> = vec![
            0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x03,
        ];
        {
            let mut b = FileBuilder::new();
            b.create_dataset("t")
                .with_raw_data(dt.clone(), raw.clone(), 3)
                .with_shape(&[3]);
            b.write(&src).unwrap();
        }

        repack(&src, &dst, &RepackOptions::new()).unwrap();

        let f = File::open(&dst).unwrap();
        let ds = f.dataset("t").unwrap();
        assert_eq!(
            ds.datatype().unwrap(),
            dt,
            "time datatype incl. byte order must survive repack"
        );
        assert_eq!(
            ds.read_raw().unwrap(),
            raw,
            "time element bytes must be preserved"
        );

        std::fs::remove_file(&src).ok();
        std::fs::remove_file(&dst).ok();
    }

    #[test]
    fn is_dropped_matches_self_and_ancestors() {
        let drop: BTreeSet<String> = ["g/old", "lone"].iter().map(|s| s.to_string()).collect();
        // The dropped path itself.
        assert!(is_dropped("lone", &drop));
        assert!(is_dropped("g/old", &drop));
        // A descendant of a dropped group is dropped (the whole subtree goes).
        assert!(is_dropped("g/old/child", &drop));
        assert!(is_dropped("g/old/a/b", &drop));
        // Unrelated paths and partial-name collisions are not dropped.
        assert!(!is_dropped("g", &drop));
        assert!(!is_dropped("g/older", &drop));
        assert!(!is_dropped("lonely", &drop));
        assert!(!is_dropped("other/old", &drop));
    }
}