osobhandClaude Opus 5.5
b086dc3c2b
gather_storage merged only runs that touch, so a strided selection of a contiguous dataset over a Storage became one range (and one owned Vec) per element: a stride-2 read of 32M f32 through File::open_storage made 16,777,232 read_at calls, took 2.0 s and peaked at 2.09 GB. The selection is now walked twice. The first walk checks the runs and plans spans: runs in increasing order at most 4 KiB apart (GATHER_GAP_BYTES) are read as one span up to 8 MiB (GATHER_SPAN_BYTES; a longer run is split), so nothing is stored per run. The spans are fetched in RAW_BATCH_BYTES batches while the second walk copies each run out of its span. Same checks and errors as before. The same read is now 32 reads and 0.31 s (File::open: 0.08 s). contiguous_read_interop: every h5py-checked selection is also read through File::open_storage and must give libhdf5's bytes; a new test bounds the range reads of strided, blocked, column and point selections (stride 2: at most 1 data read; 563,200 before). Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
489 lines
17 KiB
Rust
489 lines
17 KiB
Rust
//! Reads of contiguous datasets — full reads and hyperslab/point selections,
|
|
//! through every typed reader — checked against h5py/libhdf5 for every
|
|
//! integer and float width, both byte orders, ranks 1 to 4, and datasets
|
|
//! larger than the huge-page threshold (4 MiB) the read buffers use.
|
|
//!
|
|
//! h5py writes the file and, for each selection, reads it with libhdf5's own
|
|
//! hyperslab/point selection (`select_hyperslab` with stride and block,
|
|
//! `select_elements`) and saves the raw bytes it gets back; the byte-level
|
|
//! [`Dataset::read_selection`] must return exactly those bytes, and the typed
|
|
//! readers the same values. Skipped when python3 with h5py is unavailable,
|
|
//! unless `CLAWHDF5_REQUIRE_INTEROP=1`.
|
|
|
|
use std::path::Path;
|
|
use std::process::Command;
|
|
use std::sync::Arc;
|
|
|
|
use clawhdf5::File;
|
|
use clawhdf5_format::selection::Selection;
|
|
use clawhdf5_format::storage::CountingStorage;
|
|
|
|
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")
|
|
}
|
|
|
|
fn python_available() -> bool {
|
|
Command::new(python())
|
|
.args(["-c", "import h5py, numpy"])
|
|
.output()
|
|
.map(|o| o.status.success())
|
|
.unwrap_or(false)
|
|
}
|
|
|
|
macro_rules! skip_if_no_python {
|
|
() => {
|
|
if !python_available() {
|
|
assert!(
|
|
!interop_required(),
|
|
"CLAWHDF5_REQUIRE_INTEROP=1 but python3 with h5py is not available"
|
|
);
|
|
eprintln!("SKIP: python3 with h5py not available");
|
|
return;
|
|
}
|
|
};
|
|
}
|
|
|
|
fn run_python(script: &str) {
|
|
let output = Command::new(python())
|
|
.args(["-c", script])
|
|
.output()
|
|
.expect("failed to run python");
|
|
assert!(
|
|
output.status.success(),
|
|
"python failed:\n{}",
|
|
String::from_utf8_lossy(&output.stderr)
|
|
);
|
|
}
|
|
|
|
/// numpy type codes, with the modulus of the value pattern: every value is an
|
|
/// integer exactly representable in the type and in every typed reader's
|
|
/// output (f16 is exact below 2048, f32 below 2^24).
|
|
const DTYPES: [(&str, i64, bool); 11] = [
|
|
("i1", 201, true),
|
|
("u1", 251, false),
|
|
("i2", 2039, true),
|
|
("u2", 2039, false),
|
|
("i4", 1_000_003, true),
|
|
("u4", 1_000_003, false),
|
|
("i8", 1_000_003, true),
|
|
("u8", 1_000_003, false),
|
|
("f2", 2039, true),
|
|
("f4", 1_000_003, true),
|
|
("f8", 1_000_003, true),
|
|
];
|
|
|
|
/// Value of element `i` of a dataset of type `code` (the same formula as the
|
|
/// Python side): a permutation-revealing pattern, centred on 0 when signed.
|
|
fn value(code: &str, i: u64) -> i64 {
|
|
let (_, m, signed) = DTYPES.iter().find(|d| d.0 == code).unwrap();
|
|
let v = ((i as i128 * 7919) % *m as i128) as i64;
|
|
if *signed { v - m / 2 } else { v }
|
|
}
|
|
|
|
const SHAPES: [&[u64]; 4] = [&[1000], &[37, 53], &[7, 11, 13], &[3, 5, 7, 9]];
|
|
|
|
/// Datasets past the 4 MiB huge-page threshold, as (type, shape).
|
|
const BIG: [(&str, [u64; 2]); 4] = [
|
|
("f4", [1100, 1024]),
|
|
("i4", [1100, 1024]),
|
|
("f8", [600, 1024]),
|
|
("i8", [600, 1024]),
|
|
];
|
|
|
|
fn datasets() -> Vec<(String, String, Vec<u64>)> {
|
|
let mut out = Vec::new();
|
|
for (code, _, _) in DTYPES {
|
|
for (tag, _) in [("le", '<'), ("be", '>')] {
|
|
for shape in SHAPES {
|
|
out.push((
|
|
format!("{code}{tag}_r{}", shape.len()),
|
|
code.to_string(),
|
|
shape.to_vec(),
|
|
));
|
|
}
|
|
}
|
|
}
|
|
for (code, shape) in BIG {
|
|
for tag in ["le", "be"] {
|
|
out.push((format!("{code}{tag}_big"), code.to_string(), shape.to_vec()));
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
fn write_file(path: &Path) {
|
|
let script = format!(
|
|
r#"
|
|
import h5py, numpy as np
|
|
M = {{'i1': 201, 'u1': 251, 'i2': 2039, 'u2': 2039, 'i4': 1000003, 'u4': 1000003,
|
|
'i8': 1000003, 'u8': 1000003, 'f2': 2039, 'f4': 1000003, 'f8': 1000003}}
|
|
def values(code, n):
|
|
v = (np.arange(n, dtype=np.int64) * 7919) % M[code]
|
|
if code[0] != 'u':
|
|
v -= M[code] // 2
|
|
return v
|
|
shapes = [(1000,), (37, 53), (7, 11, 13), (3, 5, 7, 9)]
|
|
big = [('f4', (1100, 1024)), ('i4', (1100, 1024)), ('f8', (600, 1024)), ('i8', (600, 1024))]
|
|
with h5py.File("{path}", "w") as f:
|
|
for code in M:
|
|
for tag, e in (('le', '<'), ('be', '>')):
|
|
for shape in shapes:
|
|
n = int(np.prod(shape))
|
|
f.create_dataset(f"{{code}}{{tag}}_r{{len(shape)}}",
|
|
data=values(code, n).astype(e + code).reshape(shape))
|
|
for code, shape in big:
|
|
for tag, e in (('le', '<'), ('be', '>')):
|
|
n = int(np.prod(shape))
|
|
f.create_dataset(f"{{code}}{{tag}}_big",
|
|
data=values(code, n).astype(e + code).reshape(shape))
|
|
"#,
|
|
path = path.display()
|
|
);
|
|
run_python(&script);
|
|
}
|
|
|
|
#[test]
|
|
fn full_reads_match_h5py_for_every_type_order_and_size() {
|
|
skip_if_no_python!();
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let path = dir.path().join("contig.h5");
|
|
write_file(&path);
|
|
let file = File::open(&path).unwrap();
|
|
for (name, code, shape) in datasets() {
|
|
let ds = file.dataset(&name).unwrap();
|
|
assert!(ds.read_raw_ref().unwrap().is_some(), "{name} is contiguous");
|
|
let n: u64 = shape.iter().product();
|
|
let want: Vec<i64> = (0..n).map(|i| value(&code, i)).collect();
|
|
assert_eq!(
|
|
ds.read_f64().unwrap(),
|
|
want.iter().map(|&v| v as f64).collect::<Vec<_>>(),
|
|
"{name} read_f64"
|
|
);
|
|
assert_eq!(
|
|
ds.read_f32().unwrap(),
|
|
want.iter().map(|&v| v as f32).collect::<Vec<_>>(),
|
|
"{name} read_f32"
|
|
);
|
|
assert_eq!(ds.read_i64().unwrap(), want, "{name} read_i64");
|
|
assert_eq!(
|
|
ds.read_i32().unwrap(),
|
|
want.iter().map(|&v| v as i32).collect::<Vec<_>>(),
|
|
"{name} read_i32"
|
|
);
|
|
// libhdf5 saturates negative values to 0 when reading as unsigned.
|
|
assert_eq!(
|
|
ds.read_u64().unwrap(),
|
|
want.iter().map(|&v| v.max(0) as u64).collect::<Vec<_>>(),
|
|
"{name} read_u64"
|
|
);
|
|
}
|
|
}
|
|
|
|
struct Rng(u64);
|
|
impl Rng {
|
|
fn next(&mut self) -> u64 {
|
|
self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15);
|
|
let mut z = self.0;
|
|
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
|
|
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
|
|
z ^ (z >> 31)
|
|
}
|
|
fn below(&mut self, n: u64) -> u64 {
|
|
self.next() % n.max(1)
|
|
}
|
|
}
|
|
|
|
/// A hyperslab from per-dimension `(start, stride, count, block)`.
|
|
fn slab(dims: &[(u64, u64, u64, u64)]) -> Selection {
|
|
Selection::Hyperslab {
|
|
start: dims.iter().map(|d| d.0).collect(),
|
|
stride: dims.iter().map(|d| d.1).collect(),
|
|
count: dims.iter().map(|d| d.2).collect(),
|
|
block: dims.iter().map(|d| d.3).collect(),
|
|
}
|
|
}
|
|
|
|
/// Selections of every shape the read paths distinguish, all valid for `dims`.
|
|
fn selections(rng: &mut Rng, dims: &[u64]) -> Vec<Selection> {
|
|
let mut out = Vec::new();
|
|
// Unit-stride box.
|
|
out.push(slab(
|
|
&dims
|
|
.iter()
|
|
.map(|&n| {
|
|
let c = 1 + rng.below(n);
|
|
(rng.below(n - c + 1), 1, c, 1)
|
|
})
|
|
.collect::<Vec<_>>(),
|
|
));
|
|
// Strided (block 1), blocked (stride > block), and adjacent blocks
|
|
// (stride == block, which reads like a box): (block, stride - block).
|
|
for (block, gap) in [(1, 1), (2, 1), (2, 0)] {
|
|
out.push(slab(
|
|
&dims
|
|
.iter()
|
|
.map(|&n| {
|
|
let b = (block + rng.below(2)).min(n);
|
|
let st = b + gap + rng.below(2) * gap;
|
|
let s = rng.below(n - b + 1);
|
|
let c = 1 + rng.below((n - s - b) / st + 1);
|
|
(s, st, c, b)
|
|
})
|
|
.collect::<Vec<_>>(),
|
|
));
|
|
}
|
|
// Whole inner rows (one run across rows), and the whole dataset.
|
|
let r0 = rng.below(dims[0]);
|
|
let mut rows = vec![(r0, 1, 1 + rng.below(dims[0] - r0), 1)];
|
|
rows.extend(dims[1..].iter().map(|&n| (0, 1, n, 1)));
|
|
out.push(slab(&rows));
|
|
out.push(slab(
|
|
&dims.iter().map(|&n| (0, 1, n, 1)).collect::<Vec<_>>(),
|
|
));
|
|
// One element.
|
|
out.push(slab(
|
|
&dims
|
|
.iter()
|
|
.map(|&n| (rng.below(n), 1, 1, 1))
|
|
.collect::<Vec<_>>(),
|
|
));
|
|
// Distinct points in no particular order.
|
|
let mut points: Vec<Vec<u64>> = Vec::new();
|
|
for _ in 0..1 + rng.below(15) {
|
|
let p: Vec<u64> = dims.iter().map(|&n| rng.below(n)).collect();
|
|
if !points.contains(&p) {
|
|
points.push(p);
|
|
}
|
|
}
|
|
out.push(Selection::Points(points));
|
|
out
|
|
}
|
|
|
|
fn join(v: &[u64]) -> String {
|
|
v.iter().map(u64::to_string).collect::<Vec<_>>().join(",")
|
|
}
|
|
|
|
/// One element of type `code`, given as its raw file-order bytes, as an
|
|
/// integer (every value in these files is one).
|
|
fn decode(code: &str, big_endian: bool, bytes: &[u8]) -> i64 {
|
|
let mut b = bytes.to_vec();
|
|
if big_endian {
|
|
b.reverse();
|
|
}
|
|
let mut w = [0u8; 8];
|
|
w[..b.len()].copy_from_slice(&b);
|
|
let u = u64::from_le_bytes(w);
|
|
match code {
|
|
"i1" => u as u8 as i8 as i64,
|
|
"i2" => u as u16 as i16 as i64,
|
|
"i4" => u as u32 as i32 as i64,
|
|
"i8" => u as i64,
|
|
"u1" | "u2" | "u4" | "u8" => u as i64,
|
|
"f2" => clawhdf5_format::float16::f16_bits_to_f32(u as u16) as i64,
|
|
"f4" => f32::from_bits(u as u32) as i64,
|
|
"f8" => f64::from_bits(u) as i64,
|
|
_ => unreachable!(),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn selection_reads_match_h5py_for_every_type_order_and_rank() {
|
|
skip_if_no_python!();
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let path = dir.path().join("contig.h5");
|
|
write_file(&path);
|
|
|
|
let mut rng = Rng(2026);
|
|
let mut cases: Vec<(String, String, Selection)> = Vec::new();
|
|
for (name, code, shape) in datasets() {
|
|
for sel in selections(&mut rng, &shape) {
|
|
cases.push((name.clone(), code.clone(), sel));
|
|
}
|
|
// Empty: a zero count, and Selection::None.
|
|
let mut empty = shape.iter().map(|&n| (0, 1, n, 1)).collect::<Vec<_>>();
|
|
empty[shape.len() - 1].2 = 0;
|
|
cases.push((name.clone(), code.clone(), slab(&empty)));
|
|
cases.push((name, code, Selection::None));
|
|
}
|
|
let mut spec = String::new();
|
|
for (k, (name, _, sel)) in cases.iter().enumerate() {
|
|
let line = match sel {
|
|
Selection::Hyperslab { count, .. } if count.contains(&0) => "N".to_string(),
|
|
Selection::Hyperslab {
|
|
start,
|
|
stride,
|
|
count,
|
|
block,
|
|
} => format!(
|
|
"H {};{};{};{}",
|
|
join(start),
|
|
join(stride),
|
|
join(count),
|
|
join(block)
|
|
),
|
|
Selection::Points(points) => format!(
|
|
"P {}",
|
|
points.iter().map(|p| join(p)).collect::<Vec<_>>().join(";")
|
|
),
|
|
Selection::None => "N".to_string(),
|
|
Selection::All => unreachable!(),
|
|
};
|
|
spec.push_str(&format!("{k} {name} {line}\n"));
|
|
}
|
|
let spec_path = dir.path().join("cases.txt");
|
|
std::fs::write(&spec_path, spec).unwrap();
|
|
run_python(&format!(
|
|
r#"
|
|
import h5py, numpy as np
|
|
with h5py.File("{path}", "r") as f:
|
|
for line in open("{spec}"):
|
|
k, name, kind, *rest = line.split()
|
|
d = f[name]
|
|
space = d.id.get_space()
|
|
if kind == 'H':
|
|
start, stride, count, block = (tuple(int(x) for x in part.split(','))
|
|
for part in rest[0].split(';'))
|
|
space.select_hyperslab(start, count, stride, block)
|
|
elif kind == 'P':
|
|
pts = np.array([[int(x) for x in p.split(',')] for p in rest[0].split(';')],
|
|
dtype=np.uint64)
|
|
space.select_elements(pts)
|
|
else:
|
|
space.select_none()
|
|
n = space.get_select_npoints()
|
|
out = np.empty(n, dtype=d.dtype)
|
|
if n:
|
|
d.id.read(h5py.h5s.create_simple((n,)), space, out)
|
|
open("{dir}/sel_" + k + ".bin", "wb").write(out.tobytes())
|
|
"#,
|
|
path = path.display(),
|
|
spec = spec_path.display(),
|
|
dir = dir.path().display(),
|
|
));
|
|
|
|
let file = File::open(&path).unwrap();
|
|
// The same file over a storage that serves only range reads: the
|
|
// selections read only their runs (see `strided_selections_over_storage_
|
|
// read_few_ranges`) and must give the same bytes.
|
|
let remote = File::open_storage(Arc::new(CountingStorage::new(
|
|
std::fs::read(&path).unwrap(),
|
|
)))
|
|
.unwrap();
|
|
for (k, (name, code, sel)) in cases.iter().enumerate() {
|
|
let ds = file.dataset(name).unwrap();
|
|
let want_bytes = std::fs::read(dir.path().join(format!("sel_{k}.bin"))).unwrap();
|
|
let rds = remote.dataset(name).unwrap();
|
|
assert!(
|
|
rds.read_selection(sel).unwrap() == want_bytes,
|
|
"{name} {sel:?}: bytes over a range storage differ from libhdf5's"
|
|
);
|
|
assert_eq!(
|
|
rds.read_f64_selection(sel).unwrap(),
|
|
ds.read_f64_selection(sel).unwrap(),
|
|
"{name} {sel:?}: f64 over a range storage"
|
|
);
|
|
assert_eq!(
|
|
rds.read_i32_selection(sel).unwrap(),
|
|
ds.read_i32_selection(sel).unwrap(),
|
|
"{name} {sel:?}: i32 over a range storage"
|
|
);
|
|
let got_bytes = ds.read_selection(sel).unwrap();
|
|
assert!(
|
|
got_bytes == want_bytes,
|
|
"{name} {sel:?}: raw bytes differ from libhdf5's ({} vs {} bytes)",
|
|
got_bytes.len(),
|
|
want_bytes.len()
|
|
);
|
|
let size = ds.raw_datatype().unwrap().type_size() as usize;
|
|
let want: Vec<i64> = want_bytes
|
|
.chunks_exact(size)
|
|
.map(|e| decode(code, name.contains("be_"), e))
|
|
.collect();
|
|
assert_eq!(
|
|
ds.read_f64_selection(sel).unwrap(),
|
|
want.iter().map(|&v| v as f64).collect::<Vec<_>>(),
|
|
"{name} {sel:?} as f64"
|
|
);
|
|
assert_eq!(
|
|
ds.read_f32_selection(sel).unwrap(),
|
|
want.iter().map(|&v| v as f32).collect::<Vec<_>>(),
|
|
"{name} {sel:?} as f32"
|
|
);
|
|
assert_eq!(
|
|
ds.read_i64_selection(sel).unwrap(),
|
|
want,
|
|
"{name} {sel:?} as i64"
|
|
);
|
|
assert_eq!(
|
|
ds.read_i32_selection(sel).unwrap(),
|
|
want.iter().map(|&v| v as i32).collect::<Vec<_>>(),
|
|
"{name} {sel:?} as i32"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Over a storage without the file in memory, a selection of a contiguous
|
|
/// dataset reads its runs merged across small gaps: a strided selection is
|
|
/// a few large reads, not one per element (563 200 for the stride-2 case
|
|
/// before), and the values are libhdf5's.
|
|
#[test]
|
|
fn strided_selections_over_storage_read_few_ranges() {
|
|
skip_if_no_python!();
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let path = dir.path().join("contig.h5");
|
|
write_file(&path);
|
|
let storage = Arc::new(CountingStorage::new(std::fs::read(&path).unwrap()));
|
|
let remote = File::open_storage(storage.clone()).unwrap();
|
|
let local = File::open(&path).unwrap();
|
|
// f4le_big is 1100 x 1024 f32: 4 KiB rows, 4.4 MB in all.
|
|
let name = "f4le_big";
|
|
let every_100th: Vec<Vec<u64>> = (0..1100u64 * 1024)
|
|
.step_by(100)
|
|
.map(|i| vec![i / 1024, i % 1024])
|
|
.collect();
|
|
let backwards: Vec<Vec<u64>> = every_100th.iter().rev().take(50).cloned().collect();
|
|
// (selection, most range reads it may take)
|
|
let cases: Vec<(Selection, u64)> = vec![
|
|
// Every other element of every row: one read of the whole dataset.
|
|
(slab(&[(0, 1, 1100, 1), (0, 2, 512, 1)]), 1),
|
|
// Blocks of 3 every 7 on every other row: rows are 4 KiB apart, so
|
|
// one read per selected row at most.
|
|
(slab(&[(0, 2, 550, 1), (1, 7, 146, 3)]), 550),
|
|
// Every 100th element, in order: 400-byte gaps, one read.
|
|
(Selection::Points(every_100th), 1),
|
|
// Points going backwards are not merged.
|
|
(Selection::Points(backwards), 50),
|
|
// A column: 4 KiB apart, merged.
|
|
(slab(&[(0, 1, 1100, 1), (5, 1, 1, 1)]), 1),
|
|
];
|
|
let ds = remote.dataset(name).unwrap();
|
|
for (sel, most) in cases {
|
|
storage.reset();
|
|
let got = ds.read_f32_selection(&sel).unwrap();
|
|
let reads = storage.reads();
|
|
assert_eq!(
|
|
got,
|
|
local
|
|
.dataset(name)
|
|
.unwrap()
|
|
.read_f32_selection(&sel)
|
|
.unwrap(),
|
|
"{sel:?}"
|
|
);
|
|
// A few reads of metadata besides the data.
|
|
assert!(reads <= most + 8, "{sel:?}: {reads} range reads");
|
|
storage.reset();
|
|
let bytes = ds.read_selection(&sel).unwrap();
|
|
assert!(
|
|
storage.reads() <= most + 8,
|
|
"{sel:?}: {} reads",
|
|
storage.reads()
|
|
);
|
|
assert_eq!(bytes.len(), got.len() * 4);
|
|
}
|
|
}
|