h5rs tools, browser reader, libhdf5 header checks, plugin filters, concurrency benchmark #14

Merged
osobh merged 60 commits from feat/p1-proof into main 2026-09-26 13:14:39 +00:00
7 changed files with 289 additions and 13 deletions
Showing only changes of commit a5bd70216c - Show all commits
+9 -5
View File
@@ -15,7 +15,7 @@ use crate::datatype::Datatype;
use crate::error::FormatError;
use crate::extensible_array::{ExtensibleArrayHeader, read_extensible_array_chunks};
use crate::filter_pipeline::FilterPipeline;
use crate::filters::{all_filters_skipped, decompress_chunk_masked};
use crate::filters::{all_filters_skipped, decompress_chunk_exact};
use crate::fixed_array::{FixedArrayHeader, read_fixed_array_chunks};
#[cfg(feature = "std")]
use std::sync::Arc;
@@ -65,12 +65,13 @@ fn decompress_all_chunks(
let raw_chunk = &file_data[c_addr..c_addr + size];
let decompressed = if let Some(pl) = pipeline {
decompress_chunk_masked(
decompress_chunk_exact(
raw_chunk,
pl,
chunk_total_bytes,
element_size,
chunk_info.filter_mask,
&chunk_info.offsets,
)?
} else {
raw_chunk.to_vec()
@@ -932,12 +933,13 @@ pub fn read_chunked_data_cached(
let cache_them = total_bytes <= cache.max_bytes();
if let Some(pl) = pipeline {
let decode = |c: &&ChunkInfo| -> Result<Vec<u8>, FormatError> {
decompress_chunk_masked(
decompress_chunk_exact(
raw_bytes(c)?,
pl,
chunk_total_bytes,
elem_size as u32,
c.filter_mask,
&c.offsets,
)
};
for batch in misses.chunks(DECODE_BATCH) {
@@ -1217,12 +1219,13 @@ pub fn read_chunked_data_sweep(
ensure_len(file_data, c_addr, size)?;
let raw_chunk = &file_data[c_addr..c_addr + size];
let dec = if let Some(pl) = pipeline {
decompress_chunk_masked(
decompress_chunk_exact(
raw_chunk,
pl,
chunk_total_bytes,
elem_size as u32,
chunk_info.filter_mask,
&coord,
)?
} else {
raw_chunk.to_vec()
@@ -1346,12 +1349,13 @@ pub fn read_chunked_data_indexed(
ensure_len(file_data, c_addr, size)?;
let raw_chunk = &file_data[c_addr..c_addr + size];
let decompressed = if let Some(pl) = pipeline {
decompress_chunk_masked(
decompress_chunk_exact(
raw_chunk,
pl,
chunk_total_bytes,
elem_size as u32,
*filter_mask,
coord,
)?
} else {
raw_chunk.to_vec()
+62 -1
View File
@@ -4,7 +4,7 @@
extern crate alloc;
#[cfg(not(feature = "std"))]
use alloc::{boxed::Box, vec, vec::Vec};
use alloc::{boxed::Box, format, vec, vec::Vec};
use crate::error::FormatError;
#[cfg(feature = "deflate")]
@@ -120,6 +120,34 @@ pub fn decompress_chunk_masked(
Ok(data)
}
/// Decode one stored chunk of a chunked dataset: [`decompress_chunk_masked`],
/// then require exactly `chunk_size` bytes (when `chunk_size` is known).
///
/// HDF5 stores every chunk at the full chunk size — edge chunks are padded
/// before they are filtered, and an edge chunk left unfiltered is written
/// full-size too — so a pipeline that decodes to fewer bytes means a
/// corrupt chunk. libhdf5 fails such a read (or returns uninitialised
/// memory); it must never read back as zeros. `coords` (the chunk's offset
/// in the dataset) is named in the error.
pub fn decompress_chunk_exact(
compressed: &[u8],
pipeline: &FilterPipeline,
chunk_size: usize,
element_size: u32,
filter_mask: u32,
coords: &[u64],
) -> Result<Vec<u8>, FormatError> {
let data =
decompress_chunk_masked(compressed, pipeline, chunk_size, element_size, filter_mask)?;
if chunk_size != 0 && data.len() != chunk_size {
return Err(FormatError::ChunkedReadError(format!(
"chunk at {coords:?} decoded to {} bytes, expected {chunk_size}",
data.len()
)));
}
Ok(data)
}
/// Apply a filter pipeline to compress a chunk.
/// Filters are applied in FORWARD order for compression.
pub fn compress_chunk(
@@ -1516,6 +1544,39 @@ fn pcodec_decompress(
#[cfg(test)]
mod tests {
/// A chunk whose pipeline decodes to fewer bytes than the chunk holds is
/// an error naming the chunk, never a short buffer the reader pads.
#[test]
fn short_decoded_chunk_is_an_error() {
let pipeline = FilterPipeline {
version: 2,
filters: vec![FilterDescription {
filter_id: FILTER_SHUFFLE,
name: None,
flags: 0,
client_data: vec![4],
}],
};
let full = [7u8; 32];
assert_eq!(
decompress_chunk_exact(&full, &pipeline, 32, 4, 0, &[8]).unwrap(),
full
);
let err = decompress_chunk_exact(&full[..16], &pipeline, 32, 4, 0, &[8, 0]).unwrap_err();
let msg = err.to_string();
assert!(
msg.contains("[8, 0]") && msg.contains("16") && msg.contains("32"),
"{msg}"
);
// Every filter skipped: the stored bytes are the chunk, still checked.
assert!(decompress_chunk_exact(&full[..16], &pipeline, 32, 4, 1, &[0]).is_err());
// Unknown chunk size: not checked.
assert_eq!(
decompress_chunk_exact(&full[..16], &pipeline, 0, 4, 0, &[0]).unwrap(),
&full[..16]
);
}
use super::*;
use crate::filter_pipeline::FilterDescription;
+25 -1
View File
@@ -113,8 +113,15 @@ fn decode_stream(
}
/// Decode a Blosc-filtered chunk: one Blosc 1 frame.
///
/// An HDF5 chunk is never empty, so a frame that decodes to nothing where
/// the chunk size is known is corrupt (libhdf5's filter fails it too).
pub(crate) fn blosc_decode(input: &[u8], ctx: &FilterContext<'_>) -> Result<Vec<u8>, FormatError> {
blosc_decompress(input, ctx.output_limit())
let out = blosc_decompress(input, ctx.output_limit())?;
if out.is_empty() && ctx.max_output != 0 {
return Err(err("empty frame for a non-empty chunk"));
}
Ok(out)
}
/// Decompress a Blosc 1 frame, refusing more than `limit` bytes of output.
@@ -606,6 +613,23 @@ mod tests {
assert!(blosc_encode(&data, &ctx0).is_err());
}
/// A frame that declares no data, for a chunk that has some.
#[test]
fn empty_frame_for_a_non_empty_chunk_is_an_error() {
let mut frame = vec![2u8, 1, 0x20, 4];
for v in [0u32, 64, 16] {
frame.extend_from_slice(&v.to_le_bytes());
}
assert_eq!(blosc_decompress(&frame, 64).unwrap(), b"");
let f = desc(vec![2, 2, 4, 64, 5, 1, 1]);
let ctx = FilterContext {
filter: &f,
element_size: 4,
max_output: 64,
};
assert!(blosc_decode(&frame, &ctx).is_err());
}
/// A frame whose header claims a compressed size smaller than the
/// header itself, not stored raw: an error, not an arithmetic overflow
/// (it panicked in debug builds).
+64 -4
View File
@@ -10,7 +10,7 @@
use crate::chunked_read::ChunkInfo;
use crate::error::FormatError;
use crate::filter_pipeline::FilterPipeline;
use crate::filters::decompress_chunk_masked;
use crate::filters::decompress_chunk_exact;
use crate::lane_partition::{self, LaneStats, PartitionStats};
/// Threshold: only use parallel decompression when chunk count exceeds this.
@@ -84,12 +84,13 @@ pub fn decompress_chunks_lane_partitioned(
}
let raw_chunk = &file_data[c_addr..c_addr + size];
let decompressed = decompress_chunk_masked(
let decompressed = decompress_chunk_exact(
raw_chunk,
pipeline,
chunk_total_bytes,
element_size,
chunk_info.filter_mask,
&chunk_info.offsets,
)?;
stats.chunks_processed += 1;
@@ -160,12 +161,13 @@ pub fn decompress_chunks_parallel(
}
let raw_chunk = &file_data[c_addr..c_addr + size];
let decompressed = decompress_chunk_masked(
let decompressed = decompress_chunk_exact(
raw_chunk,
pipeline,
chunk_total_bytes,
element_size,
chunk_info.filter_mask,
&chunk_info.offsets,
)?;
Ok(DecompressedChunk {
@@ -204,12 +206,13 @@ pub fn decompress_chunks_sequential(
let raw_chunk = &file_data[c_addr..c_addr + size];
let decompressed = if let Some(pl) = pipeline {
decompress_chunk_masked(
decompress_chunk_exact(
raw_chunk,
pl,
chunk_total_bytes,
element_size,
chunk_info.filter_mask,
&chunk_info.offsets,
)?
} else {
raw_chunk.to_vec()
@@ -218,3 +221,60 @@ pub fn decompress_chunks_sequential(
}
Ok(result)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::filter_pipeline::{FILTER_SHUFFLE, FilterDescription};
/// Eight shuffled 32-byte chunks; chunk 5 is stored short when `short`.
fn chunks(short: bool) -> (Vec<u8>, Vec<ChunkInfo>) {
let mut file = Vec::new();
let mut infos = Vec::new();
for i in 0..8u64 {
let len = if short && i == 5 { 16 } else { 32 };
infos.push(ChunkInfo {
chunk_size: len as u32,
filter_mask: 0,
offsets: vec![i * 8],
address: file.len() as u64,
});
file.extend(core::iter::repeat_n(i as u8, len));
}
(file, infos)
}
/// Every parallel decoder refuses a chunk that decodes short, naming it.
#[test]
fn short_decoded_chunk_is_an_error() {
let pipeline = FilterPipeline {
version: 2,
filters: vec![FilterDescription {
filter_id: FILTER_SHUFFLE,
name: None,
flags: 0,
client_data: vec![4],
}],
};
let (file, good) = chunks(false);
assert_eq!(
decompress_chunks_parallel(&file, &good, &pipeline, 32, 4).unwrap()[5],
[5u8; 32]
);
let (file, bad) = chunks(true);
let errs = [
decompress_chunks_lane_partitioned(&file, &bad, &pipeline, 32, 4, 1, Some(3))
.map(|_| ())
.unwrap_err(),
decompress_chunks_parallel(&file, &bad, &pipeline, 32, 4)
.map(|_| ())
.unwrap_err(),
decompress_chunks_sequential(&file, &bad, Some(&pipeline), 32, 4)
.map(|_| ())
.unwrap_err(),
];
for e in errs {
assert!(e.to_string().contains("[40]"), "{e}");
}
}
}
+3 -2
View File
@@ -22,7 +22,7 @@ use crate::data_read::extract_selection_from_buffer;
use crate::dataspace::Dataspace;
use crate::error::FormatError;
use crate::filter_pipeline::FilterPipeline;
use crate::filters::{all_filters_skipped, decompress_chunk_masked};
use crate::filters::{all_filters_skipped, decompress_chunk_exact};
use crate::selection::Selection;
/// The smallest axis-aligned box containing every selected element, as
@@ -330,12 +330,13 @@ pub fn read_selection(
let decoded;
let data: &[u8] = match pipeline {
Some(pl) if !all_filters_skipped(pl, chunk.filter_mask) => {
decoded = decompress_chunk_masked(
decoded = decompress_chunk_exact(
raw,
pl,
chunk_bytes,
elem_size as u32,
chunk.filter_mask,
&chunk.offsets[..rank],
)?;
&decoded
}
@@ -343,3 +343,77 @@ print("OK")
let want: Vec<u8> = line[990..].iter().flat_map(|v| v.to_le_bytes()).collect();
assert_eq!(tail, want);
}
// ---------------------------------------------------------------------------
// Chunks that decode short
// ---------------------------------------------------------------------------
/// HDF5 stores every chunk at the full chunk size, so a chunk whose filters
/// decode to fewer bytes is corrupt. libhdf5 returns the rest of such a
/// chunk uninitialised (or, for filters that check, fails); clawhdf5 padded
/// it with zeros and returned it as data. Every read path must fail,
/// naming the chunk, and chunks that decode fully must still read.
#[test]
fn h5py_short_decoded_chunk_is_an_error() {
skip_if_no_python!();
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("short.h5");
let p = path.display().to_string();
run_python(&format!(
r#"
import h5py, numpy as np, zlib
with h5py.File("{p}", "w") as f:
# 1-D, 4 gzip chunks of 8 i4; chunk 2 (offset 16) inflates to 16 bytes.
ds = f.create_dataset("line", shape=(32,), chunks=(8,), dtype="<i4", compression="gzip")
ds[...] = np.arange(32, dtype="<i4") + 1
ds.id.write_direct_chunk((16,), zlib.compress(np.arange(4, dtype="<i4").tobytes()))
# 2-D with a partial edge chunk; the short chunk is the edge one (8, 4).
g = f.create_dataset("grid", shape=(10, 7), chunks=(4, 4), dtype="<f8", compression="gzip")
g[...] = np.arange(70, dtype="<f8").reshape(10, 7) + 1
g.id.write_direct_chunk((8, 4), zlib.compress(np.ones(3, dtype="<f8").tobytes()))
# 40 chunks (enough for the parallel decoder), one short, shuffle + gzip.
m = f.create_dataset("many", shape=(320,), chunks=(8,), dtype="<i4",
compression="gzip", shuffle=True)
m[...] = np.arange(320, dtype="<i4") + 1
m.id.write_direct_chunk((200,), zlib.compress(bytes(31)))
"#
));
let hyperslab = |start: u64, count: u64| Selection::Hyperslab {
start: vec![start],
stride: vec![1],
count: vec![1],
block: vec![count],
};
let file = File::open(&path).unwrap();
for (name, coords) in [("line", "[16"), ("grid", "[8, 4"), ("many", "[200")] {
let ds = file.dataset(name).unwrap();
let full = || {
if name == "grid" {
ds.read_f64().map(|_| ())
} else {
ds.read_i32().map(|_| ())
}
};
let err = full().expect_err(name).to_string();
assert!(err.contains(coords), "{name}: {err}");
// Cached reads decode the same way; a second read fails too.
assert!(full().is_err(), "{name}");
assert!(ds.read_selection(&Selection::All).is_err(), "{name}");
}
let line = file.dataset("line").unwrap();
assert!(line.read_selection(&hyperslab(14, 4)).is_err());
// A selection that avoids the short chunk still reads.
let want: Vec<u8> = (1..=16i32).flat_map(i32::to_le_bytes).collect();
assert_eq!(line.read_selection(&hyperslab(0, 16)).unwrap(), want);
let many = file.dataset("many").unwrap();
assert!(many.read_selection(&hyperslab(190, 20)).is_err());
// The memory-mapped and lazy readers.
let mm = clawhdf5::MmapFile::open(&path).unwrap();
assert!(mm.dataset("line").unwrap().read_i32().is_err());
assert!(mm.dataset("many").unwrap().read_i32().is_err());
let lazy = clawhdf5::LazyFile::open_mmap(&path).unwrap();
assert!(lazy.dataset("line").unwrap().read_i32().is_err());
assert!(lazy.dataset("grid").unwrap().read_f64().is_err());
}
@@ -415,3 +415,55 @@ with h5py.File(sys.argv[1], 'w') as f:
);
}
}
/// A corrupt chunk must never read as zeros: a Blosc frame that declares no
/// data (libhdf5's filter fails it), and Blosc, LZF and bzip2 chunks that
/// decode short (libhdf5 returns the rest of the chunk uninitialised).
#[test]
fn short_decoding_chunks_are_errors() {
if !have_python("h5py, hdf5plugin") {
return;
}
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("short.h5");
run_python(
r#"
import sys, struct, bz2
import numpy as np, h5py, hdf5plugin
def lzf(b):
# One literal run per 32 bytes: a valid LZF stream.
return b"".join(bytes([len(b[i:i+32]) - 1]) + b[i:i+32] for i in range(0, len(b), 32))
with h5py.File(sys.argv[1], 'w') as f:
kw = dict(shape=(16,), dtype='<i4', chunks=(16,))
empty = f.create_dataset('blosc_empty', **kw, **hdf5plugin.Blosc(cname='lz4'))
empty.id.write_direct_chunk((0,), bytes([2, 1, 0x20, 4]) + struct.pack('<III', 0, 64, 16))
short = f.create_dataset('blosc_short', **kw, **hdf5plugin.Blosc(cname='lz4'))
short.id.write_direct_chunk((0,), bytes([2, 1, 0x22, 4]) + struct.pack('<III', 32, 32, 48) + bytes(32))
try:
f['blosc_empty'][...]
except OSError:
pass
else:
raise SystemExit('blosc_empty: libhdf5 read it')
lz = f.create_dataset('lzf_short', **kw, compression='lzf')
lz.id.write_direct_chunk((0,), lzf(np.arange(8, dtype='<i4').tobytes()))
bz = f.create_dataset('bzip2_short', **kw, **hdf5plugin.BZip2())
bz.id.write_direct_chunk((0,), bz2.compress(np.arange(8, dtype='<i4').tobytes()))
ok = f.create_dataset('lzf_ok', **kw, compression='lzf')
ok.id.write_direct_chunk((0,), lzf(np.arange(16, dtype='<i4').tobytes()))
"#,
&[path.to_str().unwrap()],
);
let file = File::open(&path).unwrap();
for name in ["blosc_empty", "blosc_short", "lzf_short", "bzip2_short"] {
let ds = file.dataset(name).unwrap();
assert!(
ds.read_i32().is_err(),
"{name}: {:?}",
ds.read_i32().unwrap()
);
assert!(ds.read_selection(&Selection::All).is_err(), "{name}");
}
let want: Vec<i32> = (0..16).collect();
assert_eq!(file.dataset("lzf_ok").unwrap().read_i32().unwrap(), want);
}