osobhandClaude Opus 5.5
3fa5ed1dda
Every raw-data path has a generic *_in core, with the &[u8] functions as thin wrappers: data_read (read_raw_data*, read_raw_data_selection, read_chunked_native), chunked_read (the v1 B-tree chunk index, list_chunks, the full, cached, sweep and indexed reads), parallel_read, partial_read, fill_value (read_full_with_fill, apply_to_unallocated_chunks; and dataset_fill_value_from_storage is now generic), vds (the virtual file through Storage, external sources still through the resolver), vl_data (VlResolver<'a, S = [u8]>, read_vl_strings_in, read_vl_bytes_in), AttributeMessage::read_vl_strings_in and provenance::verify_dataset_in. With the whole file in memory nothing changes: chunks and contiguous data are sliced from it as before. Otherwise a chunked read lists its chunks, fetches their stored bytes with one Storage::read_ranges call per 64 MiB batch (chunks the cache already holds are not fetched), then decodes as today; a selection fetches only the chunks it overlaps, and a contiguous selection only its runs. Each extent's bounds error is the one the slice code gave, reported when that extent is reached, so errors keep their order. Tests: the equivalence harness now reads every dataset's values (whole, fill-aware, cached, indexed, three selections, VDS, VL strings and sequences) through the read_at-only storage and requires the slice results (all 653 corpus files agree); a misbehaving storage (a failing Nth read, short reads) only ever yields errors or the right values; and chunked reads are checked to use one read_ranges call. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
1206 lines
48 KiB
Rust
1206 lines
48 KiB
Rust
//! Equivalence harness for the range-read migration
|
|
//! (`docs/design/range-reads.md`, milestones M1 and M2).
|
|
//!
|
|
//! Every parser converted to [`Storage`] must give exactly what its `&[u8]`
|
|
//! form gives. This walks real files — the fixtures, files h5py writes to
|
|
//! exercise the less common structures, and optionally the conformance
|
|
//! corpus — and, for every object, runs each converted parser twice: over
|
|
//! the file as a slice, and over a [`CountingStorage`] that serves the same
|
|
//! bytes through `read_at` only (`as_contiguous()` is `None`, so no parser
|
|
//! can fall back to the whole slice). The results must be identical, value
|
|
//! for value and error for error.
|
|
//!
|
|
//! Nothing may answer [`FormatError::ContiguousStorageRequired`] any more:
|
|
//! since milestone M2 every read path works over `read_at` alone, the v2
|
|
//! B-tree structures (dense groups and attributes, a SOHM B-tree index,
|
|
//! huge fractal-heap objects) included.
|
|
//!
|
|
//! - `CLAWHDF5_STORAGE_CORPUS=dir[:dir...]` adds every `.h5`/`.hdf5`/`.he5`/
|
|
//! `.nc`/`.h5ad` file under those directories (the conformance corpus is
|
|
//! `conformance/.cache/corpus`); `CLAWHDF5_STORAGE_REPORT=1` prints the
|
|
//! per-file read counts.
|
|
//! - The h5py-written files honour `CLAWHDF5_PYTHON` and
|
|
//! `CLAWHDF5_REQUIRE_INTEROP` like the facade's interop tests.
|
|
|
|
use std::collections::{HashSet, VecDeque};
|
|
use std::fmt::Debug;
|
|
use std::path::{Path, PathBuf};
|
|
use std::process::Command;
|
|
|
|
use clawhdf5_format::attribute::{
|
|
extract_attributes_full, extract_attributes_full_in, extract_attributes_tolerant,
|
|
extract_attributes_tolerant_in,
|
|
};
|
|
use clawhdf5_format::attribute_info::AttributeInfoMessage;
|
|
use clawhdf5_format::btree_v1::{collect_symbol_table_nodes, collect_symbol_table_nodes_in};
|
|
use clawhdf5_format::btree_v2::{
|
|
BTreeV2Header, collect_btree_v2_records, collect_btree_v2_records_in, find_btree_v2_records,
|
|
find_btree_v2_records_in,
|
|
};
|
|
use clawhdf5_format::chunk_cache::ChunkCache;
|
|
use clawhdf5_format::chunked_read::{list_chunks, list_chunks_in};
|
|
use clawhdf5_format::data_layout::DataLayout;
|
|
use clawhdf5_format::data_read::{
|
|
read_raw_data_cached, read_raw_data_cached_in, read_raw_data_full, read_raw_data_full_in,
|
|
read_raw_data_indexed, read_raw_data_indexed_in, read_raw_data_selection,
|
|
read_raw_data_selection_in,
|
|
};
|
|
use clawhdf5_format::dataspace::Dataspace;
|
|
use clawhdf5_format::datatype::Datatype;
|
|
use clawhdf5_format::error::FormatError;
|
|
use clawhdf5_format::extensible_array::{
|
|
ExtensibleArrayHeader, read_extensible_array_chunks, read_extensible_array_chunks_in,
|
|
};
|
|
use clawhdf5_format::fill_value::{
|
|
dataset_fill_value_from_storage, dataset_fill_value_in, read_full_with_fill,
|
|
read_full_with_fill_in,
|
|
};
|
|
use clawhdf5_format::filter_pipeline::FilterPipeline;
|
|
use clawhdf5_format::fixed_array::{
|
|
FixedArrayHeader, read_fixed_array_chunks, read_fixed_array_chunks_in,
|
|
};
|
|
use clawhdf5_format::fractal_heap::FractalHeapHeader;
|
|
use clawhdf5_format::group_v2;
|
|
use clawhdf5_format::link_info::LinkInfoMessage;
|
|
use clawhdf5_format::local_heap::LocalHeap;
|
|
use clawhdf5_format::message_type::MessageType;
|
|
use clawhdf5_format::object_header::ObjectHeader;
|
|
use clawhdf5_format::selection::Selection;
|
|
use clawhdf5_format::shared_message::{
|
|
self, load_sohm_table, load_sohm_table_in, message_data_with_sohm, message_data_with_sohm_in,
|
|
parse_sohm_btree_entries, parse_sohm_btree_entries_in, parse_sohm_list, parse_sohm_list_in,
|
|
};
|
|
use clawhdf5_format::signature::split_user_block;
|
|
use clawhdf5_format::storage::{CountingStorage, Storage};
|
|
use clawhdf5_format::superblock::Superblock;
|
|
use clawhdf5_format::superblock_ext::{
|
|
cache_image_state, cache_image_state_in, read_superblock_extension,
|
|
read_superblock_extension_in,
|
|
};
|
|
use clawhdf5_format::symbol_table::{SymbolTableMessage, SymbolTableNode};
|
|
use clawhdf5_format::vds::{
|
|
read_virtual_dataset, read_virtual_dataset_in, virtual_dataset_extent,
|
|
virtual_dataset_extent_in,
|
|
};
|
|
use clawhdf5_format::vl_data::{VlResolver, read_vl_bytes, read_vl_bytes_in};
|
|
|
|
/// Objects visited per file, heap objects read per heap: enough to cover
|
|
/// every structure kind while keeping a 35 000-group file fast.
|
|
const MAX_OBJECTS: usize = 1500;
|
|
const MAX_HEAP_IDS: usize = 200;
|
|
/// Datasets larger than this are not read (their chunk indexes still are).
|
|
const MAX_DATA_BYTES: u64 = 16 << 20;
|
|
|
|
#[derive(Default, Debug)]
|
|
struct Tally {
|
|
files: usize,
|
|
objects: usize,
|
|
checks: usize,
|
|
reads: u64,
|
|
bytes: u64,
|
|
/// Chunk indexes (fixed and extensible arrays) read, and the most bytes
|
|
/// one of them took through the storage.
|
|
chunk_indexes: usize,
|
|
max_chunk_index_bytes: u64,
|
|
}
|
|
|
|
struct Walk<'a> {
|
|
slice: &'a [u8],
|
|
storage: &'a CountingStorage,
|
|
name: String,
|
|
/// The file's directory, for external VDS sources.
|
|
dir: Option<PathBuf>,
|
|
tally: &'a mut Tally,
|
|
}
|
|
|
|
impl Walk<'_> {
|
|
/// The storage result must equal the slice result; no parser may ask
|
|
/// for the whole file.
|
|
fn same<T: Debug>(
|
|
&mut self,
|
|
what: &str,
|
|
want: &Result<T, FormatError>,
|
|
got: &Result<T, FormatError>,
|
|
) {
|
|
self.tally.checks += 1;
|
|
if let Err(FormatError::ContiguousStorageRequired(site)) = got {
|
|
panic!(
|
|
"{}: {what} fell back to the whole file ({site}); every read path \
|
|
must work through read_at",
|
|
self.name
|
|
);
|
|
}
|
|
let (w, g) = (format!("{want:?}"), format!("{got:?}"));
|
|
assert!(
|
|
w == g,
|
|
"{}: {what} differs\n slice: {}\n storage: {}",
|
|
self.name,
|
|
&w[..w.len().min(600)],
|
|
&g[..g.len().min(600)]
|
|
);
|
|
}
|
|
|
|
fn index_read(&mut self, bytes_before: u64) {
|
|
self.tally.chunk_indexes += 1;
|
|
let bytes = self.storage.bytes_read() - bytes_before;
|
|
self.tally.max_chunk_index_bytes = self.tally.max_chunk_index_bytes.max(bytes);
|
|
}
|
|
|
|
fn st(&self) -> &dyn Storage {
|
|
self.storage
|
|
}
|
|
|
|
fn run(&mut self) {
|
|
let slice = self.slice;
|
|
let sb = Superblock::parse(slice, 0);
|
|
self.same("superblock", &sb, &Superblock::parse_in(self.st(), 0));
|
|
let Ok(sb) = sb else { return };
|
|
let (os, ls) = (sb.offset_size, sb.length_size);
|
|
|
|
let want = read_superblock_extension(slice, &sb);
|
|
self.same(
|
|
"superblock extension",
|
|
&want,
|
|
&read_superblock_extension_in(self.st(), &sb),
|
|
);
|
|
let want = cache_image_state(slice, &sb);
|
|
self.same("cache image", &want, &cache_image_state_in(self.st(), &sb));
|
|
let table = load_sohm_table(slice, os, ls);
|
|
self.same("SOHM table", &table, &load_sohm_table_in(self.st(), os, ls));
|
|
if let Ok(Some(table)) = &table {
|
|
for idx in &table.indexes {
|
|
if idx.index_type == 0 {
|
|
let want =
|
|
parse_sohm_list(slice, idx.index_addr as usize, idx.num_messages, os);
|
|
let got = parse_sohm_list_in(self.st(), idx.index_addr, idx.num_messages, os);
|
|
self.same("SOHM list", &want, &got);
|
|
} else {
|
|
let want = parse_sohm_btree_entries(slice, idx.index_addr as usize, os, ls);
|
|
let got = parse_sohm_btree_entries_in(self.st(), idx.index_addr, os, ls);
|
|
self.same("SOHM B-tree", &want, &got);
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut seen = HashSet::new();
|
|
let mut queue = VecDeque::from([sb.root_group_address]);
|
|
while let Some(addr) = queue.pop_front() {
|
|
if seen.len() >= MAX_OBJECTS || !seen.insert(addr) {
|
|
continue;
|
|
}
|
|
self.tally.objects += 1;
|
|
self.check_object(&sb, addr);
|
|
let children = group_v2::resolve_group_children(slice, &sb, addr);
|
|
let got = group_v2::resolve_group_children_in(self.st(), &sb, addr);
|
|
self.same("group listing", &children, &got);
|
|
if let Ok(children) = children {
|
|
for c in children.iter().take(MAX_HEAP_IDS) {
|
|
let want = group_v2::resolve_child(slice, &sb, addr, &c.name);
|
|
let got = group_v2::resolve_child_in(self.st(), &sb, addr, &c.name);
|
|
self.same("child lookup", &want, &got);
|
|
}
|
|
// A name no group has: the lookup's not-found path.
|
|
let want = group_v2::resolve_child(slice, &sb, addr, "no such child");
|
|
let got = group_v2::resolve_child_in(self.st(), &sb, addr, "no such child");
|
|
self.same("child lookup (missing)", &want, &got);
|
|
queue.extend(children.iter().map(|c| c.object_header_address));
|
|
}
|
|
}
|
|
// Paths: every listed name from the root, and one that is missing.
|
|
if let Ok(children) = group_v2::resolve_group_children(slice, &sb, sb.root_group_address) {
|
|
for c in children.iter().take(MAX_HEAP_IDS) {
|
|
let path = format!("/{}", c.name);
|
|
let want = group_v2::resolve_path_any(slice, &sb, &path);
|
|
let got = group_v2::resolve_path_any_in(self.st(), &sb, &path);
|
|
self.same("path", &want, &got);
|
|
}
|
|
}
|
|
let want = group_v2::resolve_path_any(slice, &sb, "/no/such/path");
|
|
let got = group_v2::resolve_path_any_in(self.st(), &sb, "/no/such/path");
|
|
self.same("path (missing)", &want, &got);
|
|
}
|
|
|
|
fn check_object(&mut self, sb: &Superblock, addr: u64) {
|
|
let slice = self.slice;
|
|
let (os, ls) = (sb.offset_size, sb.length_size);
|
|
let header = ObjectHeader::parse(slice, addr as usize, os, ls);
|
|
self.same(
|
|
"object header",
|
|
&header,
|
|
&ObjectHeader::parse_in(self.st(), addr, os, ls),
|
|
);
|
|
let Ok(header) = header else { return };
|
|
|
|
let want = extract_attributes_full(slice, &header, os, ls);
|
|
self.same(
|
|
"attributes",
|
|
&want,
|
|
&extract_attributes_full_in(self.st(), &header, os, ls),
|
|
);
|
|
let want = extract_attributes_tolerant(slice, &header, os, ls);
|
|
let got = extract_attributes_tolerant_in(self.st(), &header, os, ls);
|
|
self.same("attributes (tolerant)", &want, &got);
|
|
let want = dataset_fill_value_in(slice, &header.messages, os, ls);
|
|
self.same(
|
|
"fill value",
|
|
&want,
|
|
&dataset_fill_value_from_storage(self.st(), &header.messages, os, ls),
|
|
);
|
|
|
|
for msg in &header.messages {
|
|
if shared_message::is_shared(msg.flags) {
|
|
let want = message_data_with_sohm(slice, msg, os, ls);
|
|
let got = message_data_with_sohm_in(self.st(), msg, os, ls);
|
|
self.same("shared message", &want, &got);
|
|
}
|
|
match msg.msg_type {
|
|
MessageType::SymbolTable => {
|
|
if let Ok(stm) = SymbolTableMessage::parse(&msg.data, os) {
|
|
self.check_v1_group(&stm, os, ls);
|
|
}
|
|
}
|
|
MessageType::LinkInfo => {
|
|
if let Ok(li) = LinkInfoMessage::parse(&msg.data, os) {
|
|
self.check_heap(
|
|
li.fractal_heap_address,
|
|
li.btree_name_index_address,
|
|
4,
|
|
os,
|
|
ls,
|
|
);
|
|
}
|
|
}
|
|
MessageType::AttributeInfo => {
|
|
if let Ok(ai) = AttributeInfoMessage::parse(&msg.data, os) {
|
|
self.check_heap(
|
|
ai.fractal_heap_address,
|
|
ai.btree_name_index_address,
|
|
0,
|
|
os,
|
|
ls,
|
|
);
|
|
}
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
self.check_layout(&header, os, ls);
|
|
self.check_data(&header, os, ls);
|
|
}
|
|
|
|
/// A symbol-table group: its local heap, B-tree, nodes and names.
|
|
fn check_v1_group(&mut self, stm: &SymbolTableMessage, os: u8, ls: u8) {
|
|
let slice = self.slice;
|
|
let heap = LocalHeap::parse(slice, stm.local_heap_address as usize, os, ls);
|
|
self.same(
|
|
"local heap",
|
|
&heap,
|
|
&LocalHeap::parse_in(self.st(), stm.local_heap_address, os, ls),
|
|
);
|
|
let nodes = collect_symbol_table_nodes(slice, stm.btree_address, os, ls);
|
|
let got = collect_symbol_table_nodes_in(self.st(), stm.btree_address, os, ls);
|
|
self.same("group B-tree", &nodes, &got);
|
|
let Ok(heap) = heap else { return };
|
|
let want = heap.validate_free_list(slice, ls);
|
|
self.same(
|
|
"local heap free list",
|
|
&want,
|
|
&heap.validate_free_list_in(self.st(), ls),
|
|
);
|
|
let Ok(nodes) = nodes else { return };
|
|
for &node in nodes.iter().take(MAX_HEAP_IDS) {
|
|
let snod = SymbolTableNode::parse(slice, node as usize, os);
|
|
self.same(
|
|
"symbol table node",
|
|
&snod,
|
|
&SymbolTableNode::parse_in(self.st(), node, os),
|
|
);
|
|
let Ok(snod) = snod else { continue };
|
|
for e in &snod.entries {
|
|
let want = heap.read_string(slice, e.link_name_offset);
|
|
self.same(
|
|
"link name",
|
|
&want,
|
|
&heap.read_string_in(self.st(), e.link_name_offset),
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A dense group's or dense attributes' fractal heap: the header, and
|
|
/// the objects its name index points at. `id_at` is where the heap ID
|
|
/// starts in a name-index record (after the hash for links).
|
|
fn check_heap(&mut self, heap: Option<u64>, index: Option<u64>, id_at: usize, os: u8, ls: u8) {
|
|
let slice = self.slice;
|
|
let Some(heap_addr) = heap else { return };
|
|
let fh = FractalHeapHeader::parse(slice, heap_addr as usize, os, ls);
|
|
self.same(
|
|
"fractal heap",
|
|
&fh,
|
|
&FractalHeapHeader::parse_in(self.st(), heap_addr, os, ls),
|
|
);
|
|
let (Ok(fh), Some(index)) = (fh, index) else {
|
|
return;
|
|
};
|
|
let bt = BTreeV2Header::parse(slice, index as usize, os, ls);
|
|
self.same(
|
|
"v2 B-tree header",
|
|
&bt,
|
|
&BTreeV2Header::parse_in(self.st(), index, os, ls),
|
|
);
|
|
let Ok(bt) = bt else { return };
|
|
let records = collect_btree_v2_records(slice, &bt, os, ls);
|
|
let got = collect_btree_v2_records_in(self.st(), &bt, os, ls);
|
|
self.same("v2 B-tree records", &records, &got);
|
|
let Ok(records) = records else { return };
|
|
// Descents to single records (by their bytes), as name lookups do.
|
|
for rec in records.iter().take(8) {
|
|
let key = rec.data.clone();
|
|
let want = find_btree_v2_records(slice, &bt, os, &mut |r| r.cmp(&key[..]));
|
|
let got = find_btree_v2_records_in(self.st(), &bt, os, &mut |r| r.cmp(&key[..]));
|
|
self.same("v2 B-tree descent", &want, &got);
|
|
}
|
|
let id_len = fh.heap_id_length as usize;
|
|
for rec in records.iter().take(MAX_HEAP_IDS) {
|
|
let Some(id) = rec.data.get(id_at..id_at + id_len) else {
|
|
continue;
|
|
};
|
|
let want = fh.read_managed_object(slice, id, os);
|
|
self.same(
|
|
"heap object",
|
|
&want,
|
|
&fh.read_managed_object_in(self.st(), id, os),
|
|
);
|
|
}
|
|
}
|
|
|
|
/// A dataset's values through every raw-data path: whole reads (plain,
|
|
/// cached, indexed, fill-aware, virtual), chunk listings, selections
|
|
/// (a box, a strided hyperslab, points), VL strings and sequences.
|
|
fn check_data(&mut self, header: &ObjectHeader, os: u8, ls: u8) {
|
|
let slice = self.slice;
|
|
let find = |t: MessageType| {
|
|
header
|
|
.messages
|
|
.iter()
|
|
.find(|m| m.msg_type == t)
|
|
.and_then(|m| shared_message::message_data_with_sohm(slice, m, os, ls).ok())
|
|
};
|
|
let (Some(dt), Some(ds), Some(dl)) = (
|
|
find(MessageType::Datatype),
|
|
find(MessageType::Dataspace),
|
|
find(MessageType::DataLayout),
|
|
) else {
|
|
return;
|
|
};
|
|
let (Ok((dt, _)), Ok(ds), Ok(dl)) = (
|
|
Datatype::parse(&dt),
|
|
Dataspace::parse(&ds, ls),
|
|
DataLayout::parse(&dl, os, ls),
|
|
) else {
|
|
return;
|
|
};
|
|
let pipeline = match find(MessageType::FilterPipeline).map(|p| FilterPipeline::parse(&p)) {
|
|
Some(Ok(p)) => Some(p),
|
|
Some(Err(_)) => return,
|
|
None => None,
|
|
};
|
|
let pl = pipeline.as_ref();
|
|
let elem = dt.type_size() as u64;
|
|
let bytes = ds
|
|
.dimensions
|
|
.iter()
|
|
.try_fold(elem, |a, &d| a.checked_mul(d));
|
|
if bytes.is_none_or(|b| b > MAX_DATA_BYTES) {
|
|
return;
|
|
}
|
|
|
|
if matches!(dl, DataLayout::Virtual { .. }) {
|
|
let resolver = self.resolver();
|
|
let r: &clawhdf5_format::vds::VdsFileResolver = &resolver;
|
|
let want = virtual_dataset_extent(slice, &dl, &ds, os, ls, Some(r));
|
|
let got = virtual_dataset_extent_in(self.st(), &dl, &ds, os, ls, Some(r));
|
|
self.same("VDS extent", &want, &got);
|
|
let want = read_virtual_dataset(slice, &dl, &ds, &dt, None, os, ls, Some(r));
|
|
let got = read_virtual_dataset_in(self.st(), &dl, &ds, &dt, None, os, ls, Some(r));
|
|
self.same("VDS read", &want, &got);
|
|
return;
|
|
}
|
|
|
|
let want = read_raw_data_full(slice, &dl, &ds, &dt, pl, os, ls);
|
|
let got = read_raw_data_full_in(self.st(), &dl, &ds, &dt, pl, os, ls);
|
|
self.same("raw data", &want, &got);
|
|
let want_fill = read_full_with_fill(
|
|
&header.messages,
|
|
slice,
|
|
&dl,
|
|
&ds,
|
|
elem as usize,
|
|
os,
|
|
ls,
|
|
|| read_raw_data_full(slice, &dl, &ds, &dt, pl, os, ls),
|
|
);
|
|
let got_fill = read_full_with_fill_in(
|
|
&header.messages,
|
|
self.st(),
|
|
&dl,
|
|
&ds,
|
|
elem as usize,
|
|
os,
|
|
ls,
|
|
|| read_raw_data_full_in(self.st(), &dl, &ds, &dt, pl, os, ls),
|
|
);
|
|
self.same("raw data with fill", &want_fill, &got_fill);
|
|
|
|
if matches!(dl, DataLayout::Chunked { .. }) {
|
|
let want = list_chunks(slice, &dl, &ds, elem as usize, os, ls);
|
|
let got = list_chunks_in(self.st(), &dl, &ds, elem as usize, os, ls);
|
|
self.same("chunk list", &want, &got);
|
|
// Through a chunk cache, twice (the second read is served from
|
|
// it), and through the indexed path.
|
|
let (c1, c2) = (ChunkCache::new(), ChunkCache::new());
|
|
for _ in 0..2 {
|
|
let want = read_raw_data_cached(slice, &dl, &ds, &dt, pl, os, ls, &c1);
|
|
let got = read_raw_data_cached_in(self.st(), &dl, &ds, &dt, pl, os, ls, &c2);
|
|
// A cache lists the chunks in hash-map order (see below).
|
|
if want.is_err() && got.is_err() {
|
|
self.tally.checks += 1;
|
|
} else {
|
|
self.same("raw data (cached)", &want, &got);
|
|
}
|
|
}
|
|
let (c1, c2) = (ChunkCache::new(), ChunkCache::new());
|
|
let want = read_raw_data_indexed(slice, &dl, &ds, &dt, pl, os, ls, &c1);
|
|
let got = read_raw_data_indexed_in(self.st(), &dl, &ds, &dt, pl, os, ls, &c2);
|
|
// The indexed path decodes chunks in hash-map order, so which
|
|
// failing chunk it reports varies between two caches (with the
|
|
// slice alone, too): only whether it fails must agree.
|
|
if want.is_err() && got.is_err() {
|
|
self.tally.checks += 1;
|
|
} else {
|
|
self.same("raw data (indexed)", &want, &got);
|
|
}
|
|
}
|
|
|
|
let dims = &ds.dimensions;
|
|
if !dims.is_empty() && dims.iter().all(|&d| d > 0) {
|
|
let rank = dims.len();
|
|
let ones = vec![1u64; rank];
|
|
let quarter = Selection::Hyperslab {
|
|
start: dims.iter().map(|&d| d / 4).collect(),
|
|
stride: ones.clone(),
|
|
count: dims.iter().map(|&d| (d / 3).max(1)).collect(),
|
|
block: ones.clone(),
|
|
};
|
|
let mut stride = ones.clone();
|
|
stride[rank - 1] = 2;
|
|
let mut count = dims.clone();
|
|
count[rank - 1] = dims[rank - 1].div_ceil(2);
|
|
let strided = Selection::Hyperslab {
|
|
start: vec![0; rank],
|
|
stride,
|
|
count,
|
|
block: ones.clone(),
|
|
};
|
|
let points = Selection::Points(vec![
|
|
dims.iter().map(|&d| d - 1).collect(),
|
|
vec![0; rank],
|
|
dims.iter().map(|&d| d / 2).collect(),
|
|
]);
|
|
for (what, sel) in [
|
|
("selection (box)", &quarter),
|
|
("selection (strided)", &strided),
|
|
("selection (points)", &points),
|
|
] {
|
|
let want = read_raw_data_selection(slice, &dl, &ds, &dt, pl, os, ls, sel);
|
|
let got = read_raw_data_selection_in(self.st(), &dl, &ds, &dt, pl, os, ls, sel);
|
|
self.same(what, &want, &got);
|
|
}
|
|
}
|
|
|
|
// Variable-length strings and sequences, resolved in the global heap.
|
|
if let (Datatype::VariableLength { base_type, .. }, Ok(raw)) = (&dt, &want) {
|
|
let n = raw.len() / clawhdf5_format::vl_data::element_size(os).max(1);
|
|
let raw = &raw[..n * clawhdf5_format::vl_data::element_size(os)];
|
|
let want = VlResolver::new(slice, os, ls).string_bytes(raw);
|
|
let got = VlResolver::new_in(self.st(), os, ls).string_bytes(raw);
|
|
self.same("VL strings", &want, &got);
|
|
let base = base_type.type_size() as usize;
|
|
let want = VlResolver::new(slice, os, ls).sequences(raw, base);
|
|
let got = VlResolver::new_in(self.st(), os, ls).sequences(raw, base);
|
|
self.same("VL sequences", &want, &got);
|
|
let want = read_vl_bytes(slice, raw, n as u64, os, ls);
|
|
let got = read_vl_bytes_in(self.st(), raw, n as u64, os, ls);
|
|
self.same("VL bytes", &want, &got);
|
|
}
|
|
}
|
|
|
|
/// External VDS source files: siblings of the file being walked.
|
|
fn resolver(&self) -> impl Fn(&str) -> Result<Option<Vec<u8>>, FormatError> + use<> {
|
|
let dir = self.dir.clone();
|
|
move |name: &str| {
|
|
let (Some(dir), false) = (dir.as_ref(), name.contains("..") || name.starts_with('/'))
|
|
else {
|
|
return Ok(None);
|
|
};
|
|
Ok(std::fs::read(dir.join(name)).ok())
|
|
}
|
|
}
|
|
|
|
/// A dataset's layout: VDS mappings, and fixed/extensible array chunk
|
|
/// indexes.
|
|
fn check_layout(&mut self, header: &ObjectHeader, os: u8, ls: u8) {
|
|
let slice = self.slice;
|
|
let find = |t: MessageType| {
|
|
header
|
|
.messages
|
|
.iter()
|
|
.find(|m| m.msg_type == t)
|
|
.and_then(|m| shared_message::message_data_with_sohm(slice, m, os, ls).ok())
|
|
};
|
|
let Some(layout) = find(MessageType::DataLayout) else {
|
|
return;
|
|
};
|
|
let Ok(layout) = DataLayout::parse(&layout, os, ls) else {
|
|
return;
|
|
};
|
|
match &layout {
|
|
DataLayout::Virtual { .. } => {
|
|
let (mut want, mut got) = (layout.clone(), layout.clone());
|
|
let w = want.resolve_vds_mappings(slice, ls).map(|()| want);
|
|
let g = got.resolve_vds_mappings_in(self.st(), ls).map(|()| got);
|
|
self.same("VDS mappings", &w, &g);
|
|
}
|
|
DataLayout::Chunked {
|
|
chunk_dimensions,
|
|
btree_address: Some(addr),
|
|
version: 4,
|
|
chunk_index_type: Some(kind @ (3 | 4)),
|
|
..
|
|
} => {
|
|
let (Some(ds), Some(dt)) =
|
|
(find(MessageType::Dataspace), find(MessageType::Datatype))
|
|
else {
|
|
return;
|
|
};
|
|
let (Ok(ds), Ok((dt, _))) = (Dataspace::parse(&ds, ls), Datatype::parse(&dt))
|
|
else {
|
|
return;
|
|
};
|
|
let rank = ds.dimensions.len();
|
|
if chunk_dimensions.len() < rank {
|
|
return;
|
|
}
|
|
let dims = &chunk_dimensions[..rank];
|
|
let max = ds.max_dimensions.as_deref();
|
|
let es = dt.type_size();
|
|
if *kind == 3 {
|
|
let h = FixedArrayHeader::parse(slice, *addr as usize, os, ls);
|
|
self.same(
|
|
"fixed array header",
|
|
&h,
|
|
&FixedArrayHeader::parse_in(self.st(), *addr, os, ls),
|
|
);
|
|
let Ok(h) = h else { return };
|
|
let want =
|
|
read_fixed_array_chunks(slice, &h, &ds.dimensions, max, dims, es, os, ls);
|
|
let before = self.storage.bytes_read();
|
|
let got = read_fixed_array_chunks_in(
|
|
self.st(),
|
|
&h,
|
|
&ds.dimensions,
|
|
max,
|
|
dims,
|
|
es,
|
|
os,
|
|
ls,
|
|
);
|
|
self.index_read(before);
|
|
self.same("fixed array chunks", &want, &got);
|
|
} else {
|
|
let h = ExtensibleArrayHeader::parse(slice, *addr as usize, os, ls);
|
|
let got = ExtensibleArrayHeader::parse_in(self.st(), *addr, os, ls);
|
|
self.same("extensible array header", &h, &got);
|
|
let Ok(h) = h else { return };
|
|
let want = read_extensible_array_chunks(
|
|
slice,
|
|
&h,
|
|
&ds.dimensions,
|
|
max,
|
|
dims,
|
|
es,
|
|
os,
|
|
ls,
|
|
);
|
|
let before = self.storage.bytes_read();
|
|
let got = read_extensible_array_chunks_in(
|
|
self.st(),
|
|
&h,
|
|
&ds.dimensions,
|
|
max,
|
|
dims,
|
|
es,
|
|
os,
|
|
ls,
|
|
);
|
|
self.index_read(before);
|
|
self.same("extensible array chunks", &want, &got);
|
|
}
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn check_file(path: &Path, tally: &mut Tally) {
|
|
let Ok(bytes) = std::fs::read(path) else {
|
|
return;
|
|
};
|
|
check_bytes_in(
|
|
&path.display().to_string(),
|
|
&bytes,
|
|
path.parent().map(Path::to_path_buf),
|
|
tally,
|
|
);
|
|
}
|
|
|
|
fn check_bytes(name: &str, bytes: &[u8], tally: &mut Tally) {
|
|
check_bytes_in(name, bytes, None, tally);
|
|
}
|
|
|
|
fn check_bytes_in(name: &str, bytes: &[u8], dir: Option<PathBuf>, tally: &mut Tally) {
|
|
let Ok((_, hdf5)) = split_user_block(bytes) else {
|
|
return;
|
|
};
|
|
let storage = CountingStorage::new(hdf5.to_vec());
|
|
let before = (tally.checks, tally.objects);
|
|
let mut walk = Walk {
|
|
slice: hdf5,
|
|
storage: &storage,
|
|
name: name.to_string(),
|
|
dir,
|
|
tally,
|
|
};
|
|
walk.run();
|
|
tally.files += 1;
|
|
tally.reads += storage.reads();
|
|
tally.bytes += storage.bytes_read();
|
|
if std::env::var("CLAWHDF5_STORAGE_REPORT").is_ok_and(|v| v == "1") {
|
|
eprintln!(
|
|
"{:>6} objects {:>7} checks {:>8} reads {:>12} bytes {}",
|
|
tally.objects - before.1,
|
|
tally.checks - before.0,
|
|
storage.reads(),
|
|
storage.bytes_read(),
|
|
name
|
|
);
|
|
}
|
|
}
|
|
|
|
fn hdf5_files(dir: &Path, out: &mut Vec<PathBuf>) {
|
|
let Ok(entries) = std::fs::read_dir(dir) else {
|
|
return;
|
|
};
|
|
for e in entries.flatten() {
|
|
let p = e.path();
|
|
if p.is_dir() {
|
|
hdf5_files(&p, out);
|
|
} else if p
|
|
.extension()
|
|
.and_then(|x| x.to_str())
|
|
.is_some_and(|x| matches!(x, "h5" | "hdf5" | "he5" | "nc" | "h5ad" | "hdf"))
|
|
{
|
|
out.push(p);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn fixtures_parse_identically_through_storage() {
|
|
let mut files = Vec::new();
|
|
hdf5_files(
|
|
&Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures"),
|
|
&mut files,
|
|
);
|
|
files.sort();
|
|
assert!(files.len() >= 40, "{} fixtures", files.len());
|
|
let mut tally = Tally::default();
|
|
for f in &files {
|
|
check_file(f, &mut tally);
|
|
}
|
|
eprintln!("fixtures: {tally:?}");
|
|
assert!(tally.objects >= 150 && tally.checks >= 1000, "{tally:?}");
|
|
// Reads really went through read_at.
|
|
assert!(tally.reads > tally.objects as u64);
|
|
}
|
|
|
|
#[test]
|
|
fn corpus_parses_identically_through_storage() {
|
|
let Ok(dirs) = std::env::var("CLAWHDF5_STORAGE_CORPUS") else {
|
|
eprintln!("CLAWHDF5_STORAGE_CORPUS not set; skipping the corpus");
|
|
return;
|
|
};
|
|
let mut files = Vec::new();
|
|
for d in std::env::split_paths(&dirs) {
|
|
hdf5_files(&d, &mut files);
|
|
}
|
|
files.sort();
|
|
let mut tally = Tally::default();
|
|
for f in &files {
|
|
check_file(f, &mut tally);
|
|
}
|
|
eprintln!("corpus: {tally:?}");
|
|
assert!(tally.files > 0);
|
|
}
|
|
|
|
/// A storage that misbehaves: fails its `fail_at`-th read (1-based, `0`
|
|
/// never), and, with `short`, serves one byte less than asked for inside
|
|
/// the file (a truncated response).
|
|
struct Adversary {
|
|
data: Vec<u8>,
|
|
reads: std::sync::atomic::AtomicUsize,
|
|
fail_at: usize,
|
|
short: bool,
|
|
}
|
|
|
|
impl Storage for Adversary {
|
|
fn read_at(&self, offset: u64, len: usize) -> Result<std::borrow::Cow<'_, [u8]>, FormatError> {
|
|
let n = self
|
|
.reads
|
|
.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
|
|
+ 1;
|
|
if n == self.fail_at {
|
|
return Err(FormatError::Storage(format!(
|
|
"injected failure of read {n}"
|
|
)));
|
|
}
|
|
let got = self.data.as_slice().read_at(offset, len)?;
|
|
let mut v = got.into_owned();
|
|
if self.short && v.len() > 1 {
|
|
v.pop();
|
|
}
|
|
Ok(std::borrow::Cow::Owned(v))
|
|
}
|
|
|
|
fn len(&self) -> u64 {
|
|
self.data.len() as u64
|
|
}
|
|
}
|
|
|
|
/// Every group listing and every dataset's values (whole, fill-aware and
|
|
/// through a selection) read through a storage that fails or serves short
|
|
/// reads: each result is an error or exactly the in-memory result, never
|
|
/// other data; and a failing read is reported as that failure.
|
|
#[test]
|
|
fn misbehaving_storage_never_returns_wrong_data() {
|
|
let dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures");
|
|
let mut files = Vec::new();
|
|
hdf5_files(&dir, &mut files);
|
|
files.sort();
|
|
let (mut compared, mut failures_seen) = (0usize, 0usize);
|
|
for path in &files {
|
|
let Ok(bytes) = std::fs::read(path) else {
|
|
continue;
|
|
};
|
|
let Ok((_, hdf5)) = split_user_block(&bytes) else {
|
|
continue;
|
|
};
|
|
let Ok(sb) = Superblock::parse(hdf5, 0) else {
|
|
continue;
|
|
};
|
|
// The whole read of every object, as one result to compare.
|
|
let everything = |file: &dyn Storage| -> Result<String, FormatError> {
|
|
let (os, ls) = (sb.offset_size, sb.length_size);
|
|
let mut out = String::new();
|
|
let mut queue = VecDeque::from([sb.root_group_address]);
|
|
let mut seen = HashSet::new();
|
|
while let Some(addr) = queue.pop_front() {
|
|
if seen.len() > 200 || !seen.insert(addr) {
|
|
continue;
|
|
}
|
|
let header = ObjectHeader::parse_in(file, addr, os, ls)?;
|
|
let find = |t: MessageType| {
|
|
header
|
|
.messages
|
|
.iter()
|
|
.find(|m| m.msg_type == t)
|
|
.map(|m| message_data_with_sohm_in(file, m, os, ls))
|
|
.transpose()
|
|
};
|
|
if let (Some(dt), Some(ds), Some(dl)) = (
|
|
find(MessageType::Datatype)?,
|
|
find(MessageType::Dataspace)?,
|
|
find(MessageType::DataLayout)?,
|
|
) {
|
|
let dt = Datatype::parse(&dt)?.0;
|
|
let ds = Dataspace::parse(&ds, ls)?;
|
|
let dl = DataLayout::parse(&dl, os, ls)?;
|
|
let pl = find(MessageType::FilterPipeline)?
|
|
.map(|p| FilterPipeline::parse(&p))
|
|
.transpose()?;
|
|
let data = read_full_with_fill_in(
|
|
&header.messages,
|
|
file,
|
|
&dl,
|
|
&ds,
|
|
dt.type_size() as usize,
|
|
os,
|
|
ls,
|
|
|| read_raw_data_full_in(file, &dl, &ds, &dt, pl.as_ref(), os, ls),
|
|
);
|
|
out.push_str(&format!("{addr}: {data:?}\n"));
|
|
if let Some(&d0) = ds.dimensions.first() {
|
|
let rank = ds.dimensions.len();
|
|
let sel = Selection::Hyperslab {
|
|
start: vec![0; rank],
|
|
stride: vec![1; rank],
|
|
count: std::iter::once(d0.div_ceil(2))
|
|
.chain(ds.dimensions[1..].iter().copied())
|
|
.collect(),
|
|
block: vec![1; rank],
|
|
};
|
|
let part = read_raw_data_selection_in(
|
|
file,
|
|
&dl,
|
|
&ds,
|
|
&dt,
|
|
pl.as_ref(),
|
|
os,
|
|
ls,
|
|
&sel,
|
|
);
|
|
out.push_str(&format!("{addr} half: {part:?}\n"));
|
|
}
|
|
}
|
|
let children = group_v2::resolve_group_children_in(file, &sb, addr);
|
|
out.push_str(&format!("{addr} children: {children:?}\n"));
|
|
if let Ok(c) = children {
|
|
queue.extend(c.iter().map(|c| c.object_header_address));
|
|
}
|
|
}
|
|
Ok(out)
|
|
};
|
|
let want = everything(&hdf5);
|
|
let counting = CountingStorage::new(hdf5.to_vec());
|
|
assert_eq!(
|
|
format!("{:?}", everything(&counting)),
|
|
format!("{want:?}"),
|
|
"{}",
|
|
path.display()
|
|
);
|
|
let total = counting.reads() as usize;
|
|
let step = (total / 25).max(1);
|
|
for fail_at in (1..=total).step_by(step) {
|
|
for short in [false, true] {
|
|
if short && fail_at != 1 {
|
|
continue;
|
|
}
|
|
let adv = Adversary {
|
|
data: hdf5.to_vec(),
|
|
reads: Default::default(),
|
|
fail_at: if short { 0 } else { fail_at },
|
|
short,
|
|
};
|
|
let got = everything(&adv);
|
|
compared += 1;
|
|
match (&got, &want) {
|
|
(Ok(g), Ok(w)) => {
|
|
// Per-object results inside may be errors; values
|
|
// that were read must be the right ones.
|
|
for (gl, wl) in g.lines().zip(w.lines()) {
|
|
if gl != wl {
|
|
assert!(
|
|
gl.contains("Err("),
|
|
"{}: fail_at {fail_at} short {short}:\n got {gl}\n want {wl}",
|
|
path.display()
|
|
);
|
|
failures_seen += 1;
|
|
// A listing that failed ends the walk
|
|
// differently from here on.
|
|
if gl.contains("children: Err(") {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
(Err(FormatError::Storage(_)), _) => failures_seen += 1,
|
|
(Err(e), Ok(_)) => panic!(
|
|
"{}: fail_at {fail_at} short {short}: {e:?} instead of a storage error",
|
|
path.display()
|
|
),
|
|
(Err(_), Err(_)) => {}
|
|
// A listing failed, so the walk never reached the
|
|
// object that fails in memory.
|
|
(Ok(g), Err(e)) => assert!(
|
|
g.contains("children: Err(Storage"),
|
|
"{}: fail_at {fail_at} short {short}: read where memory fails ({e:?})",
|
|
path.display()
|
|
),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
eprintln!("misbehaving storage: {compared} runs, {failures_seen} failures reported");
|
|
assert!(
|
|
compared > 500 && failures_seen > 100,
|
|
"{compared} {failures_seen}"
|
|
);
|
|
}
|
|
|
|
/// A read_at-only storage that also counts `read_ranges` calls and ranges.
|
|
struct BatchCounting {
|
|
inner: CountingStorage,
|
|
batches: std::sync::atomic::AtomicUsize,
|
|
ranges: std::sync::atomic::AtomicUsize,
|
|
}
|
|
|
|
impl Storage for BatchCounting {
|
|
fn read_at(&self, offset: u64, len: usize) -> Result<std::borrow::Cow<'_, [u8]>, FormatError> {
|
|
self.inner.read_at(offset, len)
|
|
}
|
|
|
|
fn len(&self) -> u64 {
|
|
self.inner.len()
|
|
}
|
|
|
|
fn read_ranges(
|
|
&self,
|
|
ranges: &[std::ops::Range<u64>],
|
|
) -> Result<Vec<std::borrow::Cow<'_, [u8]>>, FormatError> {
|
|
use std::sync::atomic::Ordering::Relaxed;
|
|
self.batches.fetch_add(1, Relaxed);
|
|
self.ranges.fetch_add(ranges.len(), Relaxed);
|
|
ranges
|
|
.iter()
|
|
.map(|r| self.inner.read_at(r.start, (r.end - r.start) as usize))
|
|
.collect()
|
|
}
|
|
}
|
|
|
|
/// A chunked read lists its chunks, then fetches all their bytes with one
|
|
/// `read_ranges` call (a remote backend coalesces and parallelises it), and
|
|
/// a selection fetches only the chunks it overlaps, in one call too.
|
|
#[test]
|
|
fn chunked_reads_fetch_their_chunks_in_one_batch() {
|
|
use std::sync::atomic::Ordering::Relaxed;
|
|
let dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures");
|
|
let mut datasets = 0;
|
|
for name in [
|
|
"chunked_large.h5",
|
|
"chunked_deflate.h5",
|
|
"chunked_2d.h5",
|
|
"v4_fixed_array.h5",
|
|
] {
|
|
let bytes = std::fs::read(dir.join(name)).unwrap();
|
|
let sb = Superblock::parse(&bytes, 0).unwrap();
|
|
let (os, ls) = (sb.offset_size, sb.length_size);
|
|
let st = BatchCounting {
|
|
inner: CountingStorage::new(bytes.clone()),
|
|
batches: Default::default(),
|
|
ranges: Default::default(),
|
|
};
|
|
for child in group_v2::resolve_group_children(&bytes, &sb, sb.root_group_address).unwrap() {
|
|
let header =
|
|
ObjectHeader::parse(&bytes, child.object_header_address as usize, os, ls).unwrap();
|
|
let msg = |t: MessageType| {
|
|
header
|
|
.messages
|
|
.iter()
|
|
.find(|m| m.msg_type == t)
|
|
.map(|m| m.data.clone())
|
|
};
|
|
let Some(dl) = msg(MessageType::DataLayout) else {
|
|
continue;
|
|
};
|
|
let dl = DataLayout::parse(&dl, os, ls).unwrap();
|
|
if !matches!(dl, DataLayout::Chunked { .. }) {
|
|
continue;
|
|
}
|
|
let dt = Datatype::parse(&msg(MessageType::Datatype).unwrap())
|
|
.unwrap()
|
|
.0;
|
|
let ds = Dataspace::parse(&msg(MessageType::Dataspace).unwrap(), ls).unwrap();
|
|
let pl = msg(MessageType::FilterPipeline).map(|p| FilterPipeline::parse(&p).unwrap());
|
|
let es = dt.type_size() as usize;
|
|
let (chunks, _) = list_chunks(&bytes, &dl, &ds, es, os, ls).unwrap();
|
|
let want = read_raw_data_full(&bytes, &dl, &ds, &dt, pl.as_ref(), os, ls).unwrap();
|
|
st.batches.store(0, Relaxed);
|
|
st.ranges.store(0, Relaxed);
|
|
let got = read_raw_data_full_in(&st, &dl, &ds, &dt, pl.as_ref(), os, ls).unwrap();
|
|
assert_eq!(got, want, "{name} {}", child.name);
|
|
assert_eq!(st.batches.load(Relaxed), 1, "{name} {}", child.name);
|
|
assert_eq!(
|
|
st.ranges.load(Relaxed),
|
|
chunks.len(),
|
|
"{name} {}",
|
|
child.name
|
|
);
|
|
// The first chunk only.
|
|
let rank = ds.dimensions.len();
|
|
let sel = Selection::Hyperslab {
|
|
start: vec![0; rank],
|
|
stride: vec![1; rank],
|
|
count: vec![1; rank],
|
|
block: vec![1; rank],
|
|
};
|
|
let want = read_raw_data_selection(&bytes, &dl, &ds, &dt, pl.as_ref(), os, ls, &sel);
|
|
st.batches.store(0, Relaxed);
|
|
st.ranges.store(0, Relaxed);
|
|
let got = read_raw_data_selection_in(&st, &dl, &ds, &dt, pl.as_ref(), os, ls, &sel);
|
|
assert_eq!(got, want, "{name} {}", child.name);
|
|
if chunks.len() > 2 {
|
|
assert_eq!(st.batches.load(Relaxed), 1, "{name} {}", child.name);
|
|
assert_eq!(st.ranges.load(Relaxed), 1, "{name} {}", child.name);
|
|
}
|
|
datasets += 1;
|
|
}
|
|
}
|
|
assert!(datasets >= 4, "{datasets}");
|
|
}
|
|
|
|
fn python() -> String {
|
|
std::env::var("CLAWHDF5_PYTHON").unwrap_or_else(|_| "python3".to_string())
|
|
}
|
|
|
|
fn interop_required() -> bool {
|
|
std::env::var("CLAWHDF5_REQUIRE_INTEROP").is_ok_and(|v| v == "1")
|
|
}
|
|
|
|
/// Files h5py writes to cover what the fixtures do not: extensible arrays
|
|
/// deep enough for super blocks and paged data blocks, paged fixed arrays,
|
|
/// big symbol-table and dense groups, dense and shared attributes, a SOHM
|
|
/// list and a SOHM B-tree, committed datatypes, and a user block.
|
|
const GENERATE: &str = r#"
|
|
import ctypes, glob, os, sys, h5py, numpy as np
|
|
out = sys.argv[1]
|
|
def p(n): return os.path.join(out, n)
|
|
|
|
with h5py.File(p('ea.h5'), 'w', libver='latest') as f:
|
|
# One unlimited dimension: extensible array. 3000 chunks reach super
|
|
# blocks; 2-element chunks of int8 keep data small.
|
|
d = f.create_dataset('ea', shape=(6000,), maxshape=(None,), chunks=(2,), dtype='i1')
|
|
d[:] = np.arange(6000) % 100
|
|
d2 = f.create_dataset('ea_deflate', shape=(4000, 3), maxshape=(None, 3), chunks=(2, 3),
|
|
dtype='f4', compression='gzip')
|
|
d2[:] = np.random.default_rng(1).random((4000, 3))
|
|
d3 = f.create_dataset('ea_sparse', shape=(100000,), maxshape=(None,), chunks=(4,), dtype='i2')
|
|
d3[0:8] = 1; d3[50000:50004] = 2; d3[99996:] = 3
|
|
fa = f.create_dataset('fa_paged', shape=(5000,), chunks=(1,), dtype='u1')
|
|
fa[::3] = 7
|
|
fa2 = f.create_dataset('fa', shape=(40, 40), chunks=(8, 8), dtype='f8', compression='gzip')
|
|
fa2[:] = 1.5
|
|
for i in range(30):
|
|
f.attrs[f'a{i}'] = np.arange(i + 1)
|
|
g = f.create_group('dense')
|
|
for i in range(300):
|
|
g.create_group(f'child{i:04d}').attrs['i'] = i
|
|
f['committed'] = np.dtype([('x', 'i4'), ('y', 'f8')])
|
|
f.create_dataset('uses_committed', shape=(3,), dtype=f['committed'])
|
|
f['uses_committed'].attrs.create('ta', data=np.zeros(2, dtype=f['committed'].dtype), dtype=f['committed'])
|
|
f.attrs['vl'] = ['alpha', 'beta', 'gamma']
|
|
|
|
with h5py.File(p('v1_groups.h5'), 'w', libver='earliest', userblock_size=512) as f:
|
|
for i in range(400):
|
|
g = f.create_group(f'g{i:04d}')
|
|
g.attrs['n'] = i
|
|
f.create_dataset('x', data=np.arange(10))
|
|
|
|
# Paged chunk indexes whose data blocks are bigger than the storage reads
|
|
# in one piece (1 MiB), with two chunks written: a fixed array of 300 000
|
|
# chunks (a 2.4 MB data block) and an extensible array grown to 1.2e9 (its
|
|
# last data block holds 131 072 chunks: over 1 MiB).
|
|
with h5py.File(p('big_paged.h5'), 'w', libver='latest') as f:
|
|
d = f.create_dataset('fa', shape=(300000,), chunks=(1,), dtype='u1')
|
|
d[5] = 1; d[250000] = 2
|
|
e = f.create_dataset('ea', shape=(1,), maxshape=(None,), chunks=(1,), dtype='u1')
|
|
e.resize((1200000000,)); e[10] = 1; e[1100000000] = 3
|
|
|
|
libs = glob.glob(os.path.join(os.path.dirname(h5py.__file__), '..', 'h5py.libs', 'libhdf5-*.so*'))
|
|
if libs:
|
|
lib = ctypes.CDLL(libs[0])
|
|
lib.H5Pset_shared_mesg_nindexes.argtypes = [ctypes.c_int64, ctypes.c_uint]
|
|
lib.H5Pset_shared_mesg_index.argtypes = [ctypes.c_int64, ctypes.c_uint, ctypes.c_uint, ctypes.c_uint]
|
|
lib.H5Pset_shared_mesg_phase_change.argtypes = [ctypes.c_int64, ctypes.c_uint, ctypes.c_uint]
|
|
for name, list_max in [('sohm_list.h5', 50), ('sohm_btree.h5', 0)]:
|
|
fcpl = h5py.h5p.create(h5py.h5p.FILE_CREATE)
|
|
assert lib.H5Pset_shared_mesg_nindexes(fcpl.id, 1) >= 0
|
|
# datatype, dataspace, fill value, filter pipeline, attribute
|
|
assert lib.H5Pset_shared_mesg_index(fcpl.id, 0, 0x02 | 0x04 | 0x08 | 0x10 | 0x20, 1) >= 0
|
|
assert lib.H5Pset_shared_mesg_phase_change(fcpl.id, list_max, 0) >= 0
|
|
fapl = h5py.h5p.create(h5py.h5p.FILE_ACCESS)
|
|
fapl.set_libver_bounds(h5py.h5f.LIBVER_LATEST, h5py.h5f.LIBVER_LATEST)
|
|
fid = h5py.h5f.create(p(name).encode(), h5py.h5f.ACC_TRUNC, fcpl=fcpl, fapl=fapl)
|
|
with h5py.File(fid) as f:
|
|
for i in range(20):
|
|
d = f.create_dataset(f'd{i}', shape=(10, i + 1), dtype='f4', fillvalue=-1.0,
|
|
chunks=(5, 1), compression='gzip')
|
|
d.attrs['units'] = 'metres per second, a long enough string to share'
|
|
d.attrs['scale'] = np.arange(20, dtype='f8')
|
|
print('ok')
|
|
"#;
|
|
|
|
#[test]
|
|
fn h5py_files_parse_identically_through_storage() {
|
|
let dir = Path::new(env!("CARGO_TARGET_TMPDIR")).join("storage_equivalence");
|
|
let _ = std::fs::remove_dir_all(&dir);
|
|
std::fs::create_dir_all(&dir).unwrap();
|
|
let out = Command::new(python())
|
|
.args(["-c", GENERATE, dir.to_str().unwrap()])
|
|
.output();
|
|
match out {
|
|
Ok(o) if o.status.success() => {}
|
|
Ok(o) if interop_required() => panic!(
|
|
"h5py generation failed:\n{}\n{}",
|
|
String::from_utf8_lossy(&o.stdout),
|
|
String::from_utf8_lossy(&o.stderr)
|
|
),
|
|
Err(e) if interop_required() => panic!("python not available: {e}"),
|
|
_ => {
|
|
eprintln!("python3 with h5py unavailable; skipping");
|
|
return;
|
|
}
|
|
}
|
|
let mut files = Vec::new();
|
|
hdf5_files(&dir, &mut files);
|
|
files.sort();
|
|
let names: Vec<_> = files
|
|
.iter()
|
|
.map(|f| f.file_name().unwrap().to_string_lossy().into_owned())
|
|
.collect();
|
|
for want in ["ea.h5", "v1_groups.h5", "big_paged.h5"] {
|
|
assert!(names.iter().any(|n| n == want), "{names:?}");
|
|
}
|
|
if interop_required() {
|
|
assert!(names.iter().any(|n| n == "sohm_btree.h5"), "{names:?}");
|
|
}
|
|
let mut tally = Tally::default();
|
|
for f in &files {
|
|
if f.ends_with("big_paged.h5") {
|
|
// Only the pages in use are read, not the whole data blocks:
|
|
// read in one piece, the fixed array took 2.4 MB and the
|
|
// extensible array 1.2 MB (its super block's page bitmap and
|
|
// block addresses are most of what remains).
|
|
let mut big = Tally::default();
|
|
check_file(f, &mut big);
|
|
eprintln!("big paged blocks: {big:?}");
|
|
assert_eq!(big.chunk_indexes, 2, "{big:?}");
|
|
assert!(big.max_chunk_index_bytes < 256 << 10, "{big:?}");
|
|
// Truncated anywhere, the page-by-page reads still agree with
|
|
// the slice reads (errors included).
|
|
let bytes = std::fs::read(f).unwrap();
|
|
for cut in (0..bytes.len()).step_by(bytes.len() / 97) {
|
|
check_bytes(
|
|
&format!("big_paged.h5 cut at {cut}"),
|
|
&bytes[..cut],
|
|
&mut big,
|
|
);
|
|
}
|
|
eprintln!("big paged blocks, truncated: {big:?}");
|
|
assert!(big.chunk_indexes > 100, "{big:?}");
|
|
}
|
|
check_file(f, &mut tally);
|
|
}
|
|
eprintln!("h5py files: {tally:?}");
|
|
assert!(tally.objects >= 700, "{tally:?}");
|
|
}
|