hdf5-pure 0.41.0

//! End-to-end tests for `File::open_streaming` (issue #27).
//!
//! Writes a real HDF5 file, then reads every dataset both via the buffered
//! `File::open` and the lazy `File::open_streaming`, asserting identical
//! results. This exercises the whole streaming stack end to end: superblock
//! detection, v2 group path resolution, object-header parsing, and contiguous /
//! Fixed-Array / Extensible-Array chunked data reads — all from a `Read + Seek`
//! source that never buffers the whole file.

use hdf5_pure::{ChunkCacheConfig, File, FileAccessProperties, FileBuilder, MetadataCacheConfig};

#[test]
fn open_streaming_matches_buffered() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("streaming.h5");

    let contig: Vec<f64> = (0..100).map(|i| i as f64 * 1.5).collect();
    let fixed_chunked: Vec<i32> = (0..1000).collect();
    let unlimited_chunked: Vec<i32> = (0..500).map(|i| i * 3).collect();
    let inner: Vec<f64> = vec![10.0, 20.0, 30.0];

    {
        let mut b = FileBuilder::new();
        b.create_dataset("contig")
            .with_f64_data(&contig)
            .with_shape(&[100]);
        // Fixed shape + chunks -> Fixed Array (or implicit) chunk index.
        b.create_dataset("fixed_chunked")
            .with_i32_data(&fixed_chunked)
            .with_shape(&[1000])
            .with_chunks(&[64]);
        // Unlimited dimension -> Extensible Array chunk index.
        b.create_dataset("unlimited_chunked")
            .with_i32_data(&unlimited_chunked)
            .with_shape(&[500])
            .with_maxshape(&[u64::MAX])
            .with_chunks(&[64]);
        // Nested group + dataset exercises v2 group path resolution.
        let mut g = b.create_group("grp");
        g.create_dataset("inner")
            .with_f64_data(&inner)
            .with_shape(&[3]);
        b.add_group(g.finish());
        b.write(&path).unwrap();
    }

    let buffered = File::open(&path).unwrap();
    let streaming = File::open_streaming(&path).unwrap();

    for name in ["contig", "fixed_chunked", "unlimited_chunked", "grp/inner"] {
        let b_shape = buffered.dataset(name).unwrap().shape().unwrap();
        let s_shape = streaming.dataset(name).unwrap().shape().unwrap();
        assert_eq!(b_shape, s_shape, "shape mismatch for {name}");
    }

    // Contiguous f64.
    assert_eq!(
        streaming.dataset("contig").unwrap().read_f64().unwrap(),
        contig
    );
    // Fixed-Array-indexed chunked i32.
    assert_eq!(
        streaming
            .dataset("fixed_chunked")
            .unwrap()
            .read_i32()
            .unwrap(),
        fixed_chunked
    );
    // Extensible-Array-indexed chunked i32.
    assert_eq!(
        streaming
            .dataset("unlimited_chunked")
            .unwrap()
            .read_i32()
            .unwrap(),
        unlimited_chunked
    );
    // Dataset reached through a nested group.
    assert_eq!(
        streaming.dataset("grp/inner").unwrap().read_f64().unwrap(),
        inner
    );

    // And the streaming reads match the buffered reads byte-for-byte.
    assert_eq!(
        buffered.dataset("contig").unwrap().read_f64().unwrap(),
        streaming.dataset("contig").unwrap().read_f64().unwrap()
    );
    assert_eq!(
        buffered
            .dataset("unlimited_chunked")
            .unwrap()
            .read_i32()
            .unwrap(),
        streaming
            .dataset("unlimited_chunked")
            .unwrap()
            .read_i32()
            .unwrap()
    );
}

#[test]
fn open_streaming_with_access_properties_reads_chunked_data() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("streaming_options.h5");
    let data: Vec<i32> = (0..256).map(|i| i * 2).collect();

    {
        let mut b = FileBuilder::new();
        b.create_dataset("chunked")
            .with_i32_data(&data)
            .with_shape(&[256])
            .with_chunks(&[32]);
        b.write(&path).unwrap();
    }

    let properties = FileAccessProperties::new()
        .with_metadata_cache(MetadataCacheConfig::new(4096).with_max_entry_bytes(512))
        .with_chunk_cache(ChunkCacheConfig::disabled());
    let file = File::open_streaming_with_options(&path, properties).unwrap();
    assert_eq!(file.access_properties(), properties);

    let dataset = file.dataset("chunked").unwrap();
    assert_eq!(dataset.read_i32().unwrap(), data);
    assert_eq!(dataset.read_i32().unwrap(), data);
}

/// Build a file of `count` small datasets, enough distinct object headers that
/// a metadata cache has something to hold and a tight budget something to evict.
fn write_many_datasets(path: &std::path::Path, count: usize) {
    let mut b = FileBuilder::new();
    let values: Vec<i32> = (0..32).collect();
    for i in 0..count {
        b.create_dataset(&format!("d{i:03}"))
            .with_i32_data(&values)
            .with_shape(&[32]);
    }
    b.write(path).unwrap();
}

/// The metadata-cache budget is a number chosen before a single read has
/// happened; these are the figures that say whether it was the right one
/// (issue #353).
#[test]
fn metadata_cache_stats_report_what_the_budget_bought() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("mdc_stats.h5");
    const DATASETS: usize = 32;
    write_many_datasets(&path, DATASETS);

    let read_every = |file: &File| {
        for i in 0..DATASETS {
            file.dataset(&format!("d{i:03}"))
                .unwrap()
                .read_i32()
                .unwrap();
        }
    };

    // No budget: nothing to report. `None` rather than an all-zero snapshot,
    // which would read as a cache that is on and doing nothing.
    let uncached = File::open_streaming(&path).unwrap();
    read_every(&uncached);
    assert_eq!(uncached.metadata_cache_stats(), None);
    // A buffered open holds the whole file already, so a metadata cache would
    // be a second copy of bytes it has.
    assert_eq!(File::open(&path).unwrap().metadata_cache_stats(), None);

    const BUDGET: usize = 1 << 20;
    let file = File::open_streaming_with_options(
        &path,
        FileAccessProperties::new().with_metadata_cache(MetadataCacheConfig::new(BUDGET)),
    )
    .unwrap();

    read_every(&file);
    let warm = file
        .metadata_cache_stats()
        .expect("a budget was set, so there is a cache to report");
    assert!(warm.misses() > 0, "the first pass is what populates it");
    assert!(warm.entries() > 0 && warm.bytes() > 0);
    assert!(warm.bytes() <= BUDGET, "the budget is a bound: {warm:?}");
    assert_eq!(
        warm.evictions(),
        0,
        "1 MiB holds this file's metadata whole: {warm:?}"
    );
    assert_eq!(
        warm.invalidations(),
        0,
        "a read-only open writes nothing to invalidate against: {warm:?}"
    );

    // The reads that fill a cache miss by definition, so a rate taken over the
    // whole run charges the steady state for the warm-up. Reset, then measure
    // the part that repeats.
    file.reset_metadata_cache_stats();
    let cleared = file.metadata_cache_stats().unwrap();
    assert_eq!(cleared.reads(), 0);
    assert_eq!(cleared.hit_rate(), None);
    assert_eq!(
        (cleared.entries(), cleared.bytes()),
        (warm.entries(), warm.bytes()),
        "a reset clears counters and evicts nothing"
    );

    read_every(&file);
    let steady = file.metadata_cache_stats().unwrap();
    assert!(steady.reads() > 0);
    assert_eq!(
        steady.hit_rate(),
        Some(1.0),
        "a second pass over the same objects re-reads the same windows, and the \
         budget held every one of them: {steady:?}"
    );
}

/// The other half of the same question: a budget too small to hold the working
/// set says so, in the one figure that means "raising this would help".
#[test]
fn a_budget_too_small_for_the_file_reports_evictions() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("mdc_tight.h5");
    const DATASETS: usize = 32;
    write_many_datasets(&path, DATASETS);

    const BUDGET: usize = 1024;
    let file = File::open_streaming_with_options(
        &path,
        FileAccessProperties::new().with_metadata_cache(MetadataCacheConfig::new(BUDGET)),
    )
    .unwrap();
    for i in 0..DATASETS {
        file.dataset(&format!("d{i:03}"))
            .unwrap()
            .read_i32()
            .unwrap();
    }

    let stats = file.metadata_cache_stats().unwrap();
    assert!(
        stats.evictions() > 0,
        "32 datasets do not fit in 1 KiB of metadata: {stats:?}"
    );
    assert!(
        stats.bytes() <= BUDGET,
        "the budget still bounds it: {stats:?}"
    );
}

/// Recursively assert the streaming backend reports the identical groups,
/// datasets, shapes, and attributes as the buffered backend for the group at
/// `path`. Returns the number of attributes compared, so a caller can confirm
/// the walk actually exercised attribute reads rather than passing trivially on
/// empty maps.
fn assert_group_parity(buffered: &File, streaming: &File, path: &str) -> usize {
    let display = if path.is_empty() { "/" } else { path };
    let bg = if path.is_empty() {
        buffered.root()
    } else {
        buffered.group(path).unwrap()
    };
    let sg = if path.is_empty() {
        streaming.root()
    } else {
        streaming.group(path).unwrap()
    };

    let b_attrs = bg.attrs().unwrap();
    let s_attrs = sg.attrs().unwrap();
    assert_eq!(b_attrs, s_attrs, "group attrs mismatch at '{display}'");
    let mut count = b_attrs.len();

    let mut b_ds = bg.datasets().unwrap();
    b_ds.sort();
    let mut s_ds = sg.datasets().unwrap();
    s_ds.sort();
    assert_eq!(b_ds, s_ds, "datasets mismatch at '{display}'");
    for name in &b_ds {
        let full = child_path(path, name);
        let bd = buffered.dataset(&full).unwrap();
        let sd = streaming.dataset(&full).unwrap();
        assert_eq!(
            bd.shape().unwrap(),
            sd.shape().unwrap(),
            "shape mismatch for '{full}'"
        );
        let bda = bd.attrs().unwrap();
        let sda = sd.attrs().unwrap();
        assert_eq!(bda, sda, "dataset attrs mismatch for '{full}'");
        count += bda.len();
    }

    let mut b_g = bg.groups().unwrap();
    b_g.sort();
    let mut s_g = sg.groups().unwrap();
    s_g.sort();
    assert_eq!(b_g, s_g, "subgroups mismatch at '{display}'");
    for name in &b_g {
        count += assert_group_parity(buffered, streaming, &child_path(path, name));
    }

    count
}

fn child_path(parent: &str, name: &str) -> String {
    if parent.is_empty() {
        name.to_string()
    } else {
        format!("{parent}/{name}")
    }
}

/// The streaming backend must resolve v1 (symbol-table) groups and read
/// compact, dense (fractal-heap), and variable-length attributes identically to
/// the buffered backend. Each fixture's expected attribute count guards against
/// a trivially-empty parity pass.
#[test]
fn streaming_matches_buffered_groups_and_attributes_across_fixtures() {
    // (fixture, expected total attributes across the whole walk)
    let cases = [
        ("two_groups.h5", 0),        // v1 groups, no attributes
        ("nested_groups.h5", 0),     // nested v1 groups
        ("simple_dataset.h5", 0),    // v1 root group + dataset
        ("attrs.h5", 4),             // v1 groups + compact attrs (root + dataset)
        ("mixed_attrs.h5", 3),       // v1 subgroup + scalar/array compact attrs
        ("vl_strings.h5", 1),        // v1 root + VL-string attr (global heap)
        ("dense_attrs.h5", 50),      // v2 dataset + dense (fractal-heap) attrs
        ("dense_attrs_root.h5", 20), // v2 root group + dense attrs
        ("v2_groups.h5", 0),         // v2 groups, no attributes
    ];

    for (fixture, expected_attrs) in cases {
        let path = format!("tests/fixtures/{fixture}");
        let buffered = File::open(&path).unwrap();
        let streaming = File::open_streaming(&path).unwrap();
        let counted = assert_group_parity(&buffered, &streaming, "");
        assert_eq!(
            counted, expected_attrs,
            "attribute count mismatch for {fixture}"
        );
    }
}