h5rs tools, browser reader, libhdf5 header checks, plugin filters, concurrency benchmark #14
@@ -37,7 +37,7 @@ harness = false
|
||||
# Deflate backend: `zlib-rs` (pure Rust) by default. `fast-deflate` selects
|
||||
# zlib-ng instead (C, built with cmake); flate2 prefers a C zlib whenever one
|
||||
# is enabled, so turning it on anywhere in the build overrides the default.
|
||||
default = ["std", "checksum", "deflate", "provenance", "zlib-rs", "system-zlib-decompress"]
|
||||
default = ["std", "checksum", "deflate", "provenance", "zlib-rs", "system-zlib-decompress", "lzf"]
|
||||
std = []
|
||||
checksum = []
|
||||
deflate = ["flate2"]
|
||||
@@ -56,6 +56,9 @@ zstd = ["dep:zstd"]
|
||||
blake3_hash = ["blake3"]
|
||||
szip = ["libaec-sys"]
|
||||
pcodec = ["dep:pco"]
|
||||
# Plugin filters, pure Rust. LZF (32000) is h5py's built-in compression; it
|
||||
# has no dependencies, so it is on by default.
|
||||
lzf = []
|
||||
|
||||
[[bench]]
|
||||
name = "parallel_decompress_bench"
|
||||
|
||||
@@ -12,7 +12,7 @@ use crate::chunk_grid::ChunkGrid;
|
||||
use crate::ea_writer;
|
||||
use crate::error::FormatError;
|
||||
use crate::filter_pipeline::{
|
||||
FILTER_DEFLATE, FILTER_FLETCHER32, FILTER_LZ4, FILTER_PCODEC, FILTER_PCODEC_NAME,
|
||||
FILTER_DEFLATE, FILTER_FLETCHER32, FILTER_LZ4, FILTER_LZF, FILTER_PCODEC, FILTER_PCODEC_NAME,
|
||||
FILTER_SHUFFLE, FILTER_ZSTD, FilterDescription, FilterPipeline,
|
||||
};
|
||||
use crate::filters::compress_chunk;
|
||||
@@ -48,6 +48,45 @@ pub struct ChunkOptions {
|
||||
/// Pcodec lossless numerical compression. Private, unregistered filter
|
||||
/// ID [`FILTER_PCODEC`] (480): only clawhdf5 can read it.
|
||||
pub pcodec: bool,
|
||||
/// A plugin compression filter (LZF, ...). Takes priority over the
|
||||
/// codecs above. Each needs its cargo feature to be written.
|
||||
pub plugin: Option<PluginFilter>,
|
||||
}
|
||||
|
||||
/// A compression filter from the common HDF5 plugin set, written in the
|
||||
/// format the libhdf5 plugin (h5py / hdf5plugin) reads.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
#[non_exhaustive]
|
||||
pub enum PluginFilter {
|
||||
/// LZF (filter 32000), h5py's built-in `compression="lzf"`. Needs the
|
||||
/// `lzf` feature.
|
||||
Lzf,
|
||||
}
|
||||
|
||||
impl PluginFilter {
|
||||
/// Whether the filter reorders bytes itself, so the automatic shuffle
|
||||
/// pre-filter would only get in its way.
|
||||
fn shuffles_itself(&self) -> bool {
|
||||
match self {
|
||||
PluginFilter::Lzf => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// The pipeline entry for this filter. `chunk_bytes` is one chunk's
|
||||
/// uncompressed size (0 if unknown).
|
||||
fn description(&self, _element_size: u32, chunk_bytes: u32) -> FilterDescription {
|
||||
match self {
|
||||
// h5py's lzf_set_local: filter version, liblzf version, chunk
|
||||
// size in bytes. Optional, as h5py flags it: a chunk the filter
|
||||
// cannot shrink may then be stored unfiltered.
|
||||
PluginFilter::Lzf => FilterDescription {
|
||||
filter_id: FILTER_LZF,
|
||||
name: Some("lzf".into()),
|
||||
flags: 1,
|
||||
client_data: vec![4, 0x0105, chunk_bytes],
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Largest chunk the automatic choice produces, in bytes.
|
||||
@@ -92,14 +131,33 @@ impl ChunkOptions {
|
||||
|| self.lz4
|
||||
|| self.zstd_level.is_some()
|
||||
|| self.pcodec
|
||||
|| self.plugin.is_some()
|
||||
}
|
||||
|
||||
/// Build a FilterPipeline from the options.
|
||||
pub fn build_pipeline(&self, element_size: u32) -> Option<FilterPipeline> {
|
||||
self.build_pipeline_for_chunk(element_size, 0)
|
||||
}
|
||||
|
||||
/// Build a FilterPipeline for chunks of `chunk_bytes` uncompressed bytes
|
||||
/// (0 if unknown). Some plugin filters record the chunk size in their
|
||||
/// client data.
|
||||
pub fn build_pipeline_for_chunk(
|
||||
&self,
|
||||
element_size: u32,
|
||||
chunk_bytes: u32,
|
||||
) -> Option<FilterPipeline> {
|
||||
let mut filters = Vec::new();
|
||||
|
||||
let has_compression =
|
||||
self.deflate_level.is_some() || self.zstd_level.is_some() || self.lz4 || self.pcodec;
|
||||
let plugin_shuffles = self
|
||||
.plugin
|
||||
.as_ref()
|
||||
.is_some_and(PluginFilter::shuffles_itself);
|
||||
let has_compression = self.deflate_level.is_some()
|
||||
|| self.zstd_level.is_some()
|
||||
|| self.lz4
|
||||
|| self.pcodec
|
||||
|| (self.plugin.is_some() && !plugin_shuffles);
|
||||
|
||||
// Shuffle before compression. Applied if explicitly requested OR if compression
|
||||
// is active and the caller hasn't disabled it — matches h5py default behavior
|
||||
@@ -113,8 +171,11 @@ impl ChunkOptions {
|
||||
});
|
||||
}
|
||||
|
||||
// Compression filters (mutually exclusive, priority: pcodec > zstd > lz4 > deflate)
|
||||
if self.pcodec {
|
||||
// Compression filters (mutually exclusive, priority: plugin > pcodec >
|
||||
// zstd > lz4 > deflate)
|
||||
if let Some(plugin) = &self.plugin {
|
||||
filters.push(plugin.description(element_size, chunk_bytes));
|
||||
} else if self.pcodec {
|
||||
filters.push(FilterDescription {
|
||||
filter_id: FILTER_PCODEC,
|
||||
name: Some(FILTER_PCODEC_NAME.into()),
|
||||
@@ -675,7 +736,12 @@ pub fn precompress_chunks(
|
||||
element_size: usize,
|
||||
options: &ChunkOptions,
|
||||
) -> Result<PrecompressedChunks, FormatError> {
|
||||
let pipeline = options.build_pipeline(element_size as u32);
|
||||
let chunk_bytes = chunk_dims
|
||||
.iter()
|
||||
.try_fold(element_size as u64, |acc, &d| acc.checked_mul(d))
|
||||
.and_then(|b| u32::try_from(b).ok())
|
||||
.unwrap_or(0);
|
||||
let pipeline = options.build_pipeline_for_chunk(element_size as u32, chunk_bytes);
|
||||
let has_filters = pipeline.is_some();
|
||||
let pipeline_message = pipeline.as_ref().map(|pl| pl.serialize());
|
||||
|
||||
@@ -1569,6 +1635,20 @@ mod tests {
|
||||
assert_eq!(pl.filters[1].client_data, vec![3]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn chunk_options_pipeline_lzf() {
|
||||
let options = ChunkOptions {
|
||||
plugin: Some(PluginFilter::Lzf),
|
||||
..Default::default()
|
||||
};
|
||||
assert!(options.is_chunked());
|
||||
let pl = options.build_pipeline_for_chunk(8, 800).unwrap();
|
||||
assert_eq!(pl.filters.len(), 2);
|
||||
assert_eq!(pl.filters[0].filter_id, FILTER_SHUFFLE);
|
||||
assert_eq!(pl.filters[1].filter_id, FILTER_LZF);
|
||||
assert_eq!(pl.filters[1].client_data, vec![4, 0x0105, 800]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn chunk_options_zstd_priority_over_deflate() {
|
||||
let options = ChunkOptions {
|
||||
|
||||
@@ -19,6 +19,18 @@ pub const FILTER_SCALEOFFSET: u16 = 6;
|
||||
pub const FILTER_LZ4: u16 = 32004;
|
||||
/// Zstandard compression.
|
||||
pub const FILTER_ZSTD: u16 = 32015;
|
||||
/// bzip2 (registered by PyTables; hdf5plugin's `BZip2`).
|
||||
pub const FILTER_BZIP2: u16 = 307;
|
||||
/// LZF — h5py's built-in `compression="lzf"`.
|
||||
pub const FILTER_LZF: u16 = 32000;
|
||||
/// Blosc 1 (hdf5-blosc; hdf5plugin's `Blosc`).
|
||||
pub const FILTER_BLOSC: u16 = 32001;
|
||||
/// Bitshuffle, optionally with LZ4 or Zstandard (hdf5plugin's `Bitshuffle`).
|
||||
pub const FILTER_BITSHUFFLE: u16 = 32008;
|
||||
/// ZFP lossy floating-point compression (hdf5plugin's `Zfp`). Not supported.
|
||||
pub const FILTER_ZFP: u16 = 32013;
|
||||
/// Blosc 2 (hdf5plugin's `Blosc2`).
|
||||
pub const FILTER_BLOSC2: u16 = 32026;
|
||||
/// Pcodec lossless numerical codec — a **private, unregistered** clawhdf5
|
||||
/// filter. Pcodec has no ID in the HDF Group's filter registry (checked
|
||||
/// 2026-09-25, `hdf5_plugins/docs/RegisteredFilterPlugins.md`), so it uses an
|
||||
|
||||
@@ -192,6 +192,13 @@ pub(crate) static BUILTIN_FILTERS: &[BuiltinFilter] = &[
|
||||
decode: |d, c| pcodec_decompress(d, c.element_size, c.max_output),
|
||||
encode: Some(|d, c| pcodec_compress(d, c.element_size)),
|
||||
},
|
||||
#[cfg(feature = "lzf")]
|
||||
BuiltinFilter {
|
||||
id: crate::filter_pipeline::FILTER_LZF,
|
||||
name: "lzf",
|
||||
decode: crate::filters_lzf::lzf_decode,
|
||||
encode: Some(crate::filters_lzf::lzf_encode),
|
||||
},
|
||||
#[cfg(feature = "lz4")]
|
||||
BuiltinFilter {
|
||||
id: FILTER_LZ4,
|
||||
@@ -1702,7 +1709,7 @@ mod tests {
|
||||
// An unsupported filter is fine when the chunk skipped it.
|
||||
let unknown = FilterPipeline {
|
||||
version: 2,
|
||||
filters: vec![filter(32000), filter(FILTER_DEFLATE)],
|
||||
filters: vec![filter(32013), filter(FILTER_DEFLATE)],
|
||||
};
|
||||
assert_eq!(
|
||||
decompress_chunk_masked(&deflated, &unknown, n, 8, 0b01).unwrap(),
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
//! LZF (HDF5 filter 32000) — h5py's built-in compression filter
|
||||
//! (`compression="lzf"`), in pure Rust.
|
||||
//!
|
||||
//! The chunk is one raw LZF stream (liblzf 3.x format, no header). The
|
||||
//! stream is a sequence of instructions, each starting with a control byte:
|
||||
//!
|
||||
//! * `000LLLLL` — a literal run: the next `L + 1` bytes (1..=32) are copied.
|
||||
//! * `LLLOOOOO [E] OOOOOOOO` — a back reference: copy `len + 2` bytes from
|
||||
//! `distance` bytes back, where `len` is the top three bits (1..=6), or
|
||||
//! `7 + E` when they are all ones, and `distance` is the 13-bit offset
|
||||
//! (high five bits in the control byte, low eight in the last byte) plus 1.
|
||||
//!
|
||||
//! h5py's filter (`lzf_filter.c`) records the chunk's size in bytes in
|
||||
//! `cd_values[2]` (slots 0 and 1 hold the filter and liblzf versions) and
|
||||
//! sizes its output buffer from it.
|
||||
|
||||
#[cfg(not(feature = "std"))]
|
||||
extern crate alloc;
|
||||
#[cfg(not(feature = "std"))]
|
||||
use alloc::{format, vec, vec::Vec};
|
||||
|
||||
use crate::error::FormatError;
|
||||
use crate::filter_registry::FilterContext;
|
||||
|
||||
/// `H5PY_FILTER_LZF_VERSION`, written to `cd_values[0]`.
|
||||
pub const LZF_FILTER_VERSION: u32 = 4;
|
||||
/// `LZF_VERSION` (liblzf 1.5), written to `cd_values[1]`.
|
||||
pub const LZF_API_VERSION: u32 = 0x0105;
|
||||
|
||||
const MAX_LITERAL: usize = 32;
|
||||
const MAX_OFFSET: usize = 1 << 13;
|
||||
const MAX_REF: usize = (1 << 8) + (1 << 3);
|
||||
const HASH_LOG: u32 = 14;
|
||||
|
||||
fn err(msg: &str) -> FormatError {
|
||||
FormatError::DecompressionError(format!("lzf: {msg}"))
|
||||
}
|
||||
|
||||
/// Decode an LZF-filtered chunk.
|
||||
pub(crate) fn lzf_decode(input: &[u8], ctx: &FilterContext<'_>) -> Result<Vec<u8>, FormatError> {
|
||||
let limit = ctx.output_limit();
|
||||
let hint = match ctx.client_data().get(2) {
|
||||
Some(&n) if n != 0 => n as usize,
|
||||
_ => input.len().saturating_mul(2),
|
||||
};
|
||||
lzf_decompress(input, hint.min(limit), limit)
|
||||
}
|
||||
|
||||
/// Decompress a raw LZF stream, refusing to produce more than `limit` bytes.
|
||||
pub fn lzf_decompress(
|
||||
input: &[u8],
|
||||
size_hint: usize,
|
||||
limit: usize,
|
||||
) -> Result<Vec<u8>, FormatError> {
|
||||
let mut out: Vec<u8> = Vec::new();
|
||||
out.try_reserve(size_hint)
|
||||
.map_err(|_| err("cannot allocate the output buffer"))?;
|
||||
let mut ip = 0usize;
|
||||
while ip < input.len() {
|
||||
let ctrl = input[ip] as usize;
|
||||
ip += 1;
|
||||
if ctrl < 32 {
|
||||
let run = ctrl + 1;
|
||||
let lit = input
|
||||
.get(ip..ip + run)
|
||||
.ok_or_else(|| err("literal run past the end of the input"))?;
|
||||
if out.len() + run > limit {
|
||||
return Err(err("output exceeds the chunk size"));
|
||||
}
|
||||
out.extend_from_slice(lit);
|
||||
ip += run;
|
||||
} else {
|
||||
let mut len = ctrl >> 5;
|
||||
if len == 7 {
|
||||
len += *input
|
||||
.get(ip)
|
||||
.ok_or_else(|| err("truncated back reference"))?
|
||||
as usize;
|
||||
ip += 1;
|
||||
}
|
||||
let low = *input
|
||||
.get(ip)
|
||||
.ok_or_else(|| err("truncated back reference"))? as usize;
|
||||
ip += 1;
|
||||
let distance = ((ctrl & 0x1f) << 8) + low + 1;
|
||||
let len = len + 2;
|
||||
if distance > out.len() {
|
||||
return Err(err("back reference before the start of the output"));
|
||||
}
|
||||
if out.len() + len > limit {
|
||||
return Err(err("output exceeds the chunk size"));
|
||||
}
|
||||
let start = out.len() - distance;
|
||||
if distance >= len {
|
||||
out.extend_from_within(start..start + len);
|
||||
} else {
|
||||
// Overlapping copy: repeats the last `distance` bytes.
|
||||
for k in 0..len {
|
||||
let b = out[start + k];
|
||||
out.push(b);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// Encode a chunk with the LZF filter.
|
||||
pub(crate) fn lzf_encode(input: &[u8], _ctx: &FilterContext<'_>) -> Result<Vec<u8>, FormatError> {
|
||||
Ok(lzf_compress(input))
|
||||
}
|
||||
|
||||
fn hash3(b: &[u8]) -> usize {
|
||||
let v = (u32::from(b[0]) << 16) | (u32::from(b[1]) << 8) | u32::from(b[2]);
|
||||
(v.wrapping_mul(2_654_435_761) >> (32 - HASH_LOG)) as usize
|
||||
}
|
||||
|
||||
fn flush_literals(out: &mut Vec<u8>, lit: &[u8]) {
|
||||
for run in lit.chunks(MAX_LITERAL) {
|
||||
out.push((run.len() - 1) as u8);
|
||||
out.extend_from_slice(run);
|
||||
}
|
||||
}
|
||||
|
||||
/// Compress `input` into a raw LZF stream any liblzf decoder reads.
|
||||
///
|
||||
/// Incompressible input grows by one byte per 32. (h5py's own filter gives
|
||||
/// up on such a chunk and stores it unfiltered; storing the slightly larger
|
||||
/// stream is equally readable.)
|
||||
pub fn lzf_compress(input: &[u8]) -> Vec<u8> {
|
||||
let n = input.len();
|
||||
let mut out = Vec::with_capacity(n + n / MAX_LITERAL + 1);
|
||||
let mut table = vec![0u32; 1 << HASH_LOG];
|
||||
let mut lit_start = 0usize;
|
||||
let mut i = 0usize;
|
||||
while i + 2 < n {
|
||||
let h = hash3(&input[i..]);
|
||||
let cand = table[h] as usize;
|
||||
table[h] = (i + 1) as u32;
|
||||
if cand != 0 {
|
||||
let r = cand - 1;
|
||||
let distance = i - r;
|
||||
if distance <= MAX_OFFSET && input[r..r + 3] == input[i..i + 3] {
|
||||
let max_len = (n - i).min(MAX_REF);
|
||||
let mut len = 3;
|
||||
while len < max_len && input[r + len] == input[i + len] {
|
||||
len += 1;
|
||||
}
|
||||
flush_literals(&mut out, &input[lit_start..i]);
|
||||
let code = len - 2;
|
||||
let off = distance - 1;
|
||||
if code < 7 {
|
||||
out.push(((code << 5) | (off >> 8)) as u8);
|
||||
} else {
|
||||
out.push(((7 << 5) | (off >> 8)) as u8);
|
||||
out.push((code - 7) as u8);
|
||||
}
|
||||
out.push((off & 0xff) as u8);
|
||||
// Index the positions the match covered so later data can
|
||||
// refer back into it.
|
||||
let end = i + len;
|
||||
let mut j = i + 1;
|
||||
while j < end && j + 2 < n {
|
||||
table[hash3(&input[j..])] = (j + 1) as u32;
|
||||
j += 1;
|
||||
}
|
||||
i = end;
|
||||
lit_start = i;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
i += 1;
|
||||
}
|
||||
flush_literals(&mut out, &input[lit_start..]);
|
||||
out
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn round_trip(data: &[u8]) {
|
||||
let c = lzf_compress(data);
|
||||
assert_eq!(lzf_decompress(&c, data.len(), data.len()).unwrap(), data);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn round_trips() {
|
||||
round_trip(b"");
|
||||
round_trip(b"a");
|
||||
round_trip(b"abcabcabcabcabcabcabcabcabcabcabcabc");
|
||||
round_trip(&[7u8; 10_000]);
|
||||
let noise: Vec<u8> = (0..70_000u32)
|
||||
.map(|i| (i.wrapping_mul(2_654_435_761) >> 13) as u8)
|
||||
.collect();
|
||||
round_trip(&noise);
|
||||
let ramp: Vec<u8> = (0..100_000u32)
|
||||
.flat_map(|i| (i % 1000).to_le_bytes())
|
||||
.collect();
|
||||
round_trip(&ramp);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn compresses_repetitive_data() {
|
||||
let data = [42u8; 4096];
|
||||
assert!(lzf_compress(&data).len() < 100);
|
||||
}
|
||||
|
||||
/// The chunk h5py 3.16's bundled liblzf writes for
|
||||
/// `b"hello hello hello hello"` (read back with `read_direct_chunk`): a
|
||||
/// 7-byte literal, a 14-byte back reference 6 bytes back (extended
|
||||
/// length), and a 2-byte literal.
|
||||
#[test]
|
||||
fn decodes_liblzf_output() {
|
||||
let stream = b"\x06hello h\xe0\x05\x05\x01lo";
|
||||
assert_eq!(
|
||||
lzf_decompress(stream, 23, 23).unwrap(),
|
||||
b"hello hello hello hello"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_corrupt_streams() {
|
||||
// Back reference before the start.
|
||||
assert!(lzf_decompress(&[0x20, 0x00], 10, 10).is_err());
|
||||
// Literal run past the end.
|
||||
assert!(lzf_decompress(&[0x05, 1, 2], 10, 10).is_err());
|
||||
// Output over the limit.
|
||||
let c = lzf_compress(&[1u8; 100]);
|
||||
assert!(lzf_decompress(&c, 10, 99).is_err());
|
||||
}
|
||||
}
|
||||
@@ -73,6 +73,8 @@ pub mod fill_value;
|
||||
pub mod filter_pipeline;
|
||||
pub mod filter_registry;
|
||||
pub mod filters;
|
||||
#[cfg(feature = "lzf")]
|
||||
pub mod filters_lzf;
|
||||
mod filters_szip;
|
||||
pub mod fixed_array;
|
||||
pub mod float16;
|
||||
|
||||
@@ -731,6 +731,24 @@ impl DatasetBuilder {
|
||||
self
|
||||
}
|
||||
|
||||
/// Compress with a plugin filter ([`PluginFilter`]), in the format the
|
||||
/// libhdf5 plugin reads (h5py, hdf5plugin). Implies chunked storage.
|
||||
/// Each filter needs its cargo feature (`lzf`, ...); writing fails with
|
||||
/// `UnsupportedFilter` without it.
|
||||
///
|
||||
/// [`PluginFilter`]: crate::chunked_write::PluginFilter
|
||||
pub fn with_plugin_filter(&mut self, filter: crate::chunked_write::PluginFilter) -> &mut Self {
|
||||
self.chunk_options.plugin = Some(filter);
|
||||
self
|
||||
}
|
||||
|
||||
/// Enable LZF compression (filter 32000) — h5py's built-in
|
||||
/// `compression="lzf"`. Implies chunked storage; shuffle is applied
|
||||
/// first unless `.without_shuffle()`. Requires the `lzf` cargo feature.
|
||||
pub fn with_lzf(&mut self) -> &mut Self {
|
||||
self.with_plugin_filter(crate::chunked_write::PluginFilter::Lzf)
|
||||
}
|
||||
|
||||
/// Enable Pcodec lossless numerical compression (private clawhdf5 filter
|
||||
/// ID 480).
|
||||
///
|
||||
|
||||
@@ -31,7 +31,7 @@ name = "parallel_bench"
|
||||
harness = false
|
||||
|
||||
[features]
|
||||
default = ["mmap", "provenance"]
|
||||
default = ["mmap", "provenance", "lzf"]
|
||||
mmap = ["clawhdf5-io/mmap"]
|
||||
parallel = ["clawhdf5-format/parallel", "rayon"]
|
||||
# zlib-ng (C, needs cmake) instead of the default pure-Rust zlib-rs.
|
||||
@@ -41,6 +41,9 @@ zstd = ["clawhdf5-format/zstd"]
|
||||
blake3_hash = ["clawhdf5-format/blake3_hash"]
|
||||
lz4 = ["clawhdf5-format/lz4"]
|
||||
pcodec = ["clawhdf5-format/pcodec"]
|
||||
# Plugin filters, pure Rust (no C). LZF (32000) is h5py's built-in
|
||||
# compression; it has no dependencies, so it is on by default.
|
||||
lzf = ["clawhdf5-format/lzf"]
|
||||
# Dataset::verify_provenance() — recompute a dataset's SHA-256 and compare
|
||||
# against its stored _provenance_sha256 attribute. On by default, matching
|
||||
# clawhdf5-format's own default-on `provenance` feature.
|
||||
|
||||
@@ -0,0 +1,263 @@
|
||||
//! Plugin filters (LZF, bitshuffle, bzip2, blosc) against libhdf5.
|
||||
//!
|
||||
//! Read direction: h5py (with hdf5plugin for everything but LZF, which h5py
|
||||
//! ships) writes each filter over a matrix of dtypes (1-8 bytes, both byte
|
||||
//! orders), 1-3 dimensional shapes whose chunks do not divide them (so the
|
||||
//! edge chunks are partial), compressible and incompressible data, and the
|
||||
//! filter's own options; every dataset has an unfiltered twin, and clawhdf5
|
||||
//! must read the filtered one byte for byte equal to it.
|
||||
//!
|
||||
//! Write direction: clawhdf5 writes with its encoder, and h5py must read the
|
||||
//! values back.
|
||||
//!
|
||||
//! Skipped when python3 with h5py (and hdf5plugin) is unavailable, unless
|
||||
//! `CLAWHDF5_REQUIRE_INTEROP=1`.
|
||||
#![allow(dead_code)]
|
||||
|
||||
use std::process::Command;
|
||||
|
||||
use clawhdf5::File;
|
||||
use clawhdf5_format::selection::Selection;
|
||||
|
||||
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_has(modules: &str) -> bool {
|
||||
Command::new(python())
|
||||
.args(["-c", &format!("import {modules}")])
|
||||
.output()
|
||||
.map(|o| o.status.success())
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
/// Whether the interop test can run; panics instead of skipping when
|
||||
/// `CLAWHDF5_REQUIRE_INTEROP=1`.
|
||||
fn have_python(modules: &str) -> bool {
|
||||
if python_has(modules) {
|
||||
return true;
|
||||
}
|
||||
assert!(
|
||||
!interop_required(),
|
||||
"CLAWHDF5_REQUIRE_INTEROP=1 but python3 with {modules} is not available"
|
||||
);
|
||||
eprintln!("SKIP: python3 with {modules} not available");
|
||||
false
|
||||
}
|
||||
|
||||
fn run_python(script: &str, args: &[&str]) -> String {
|
||||
let output = Command::new(python())
|
||||
.arg("-c")
|
||||
.arg(script)
|
||||
.args(args)
|
||||
.output()
|
||||
.expect("failed to run python");
|
||||
if !output.status.success() {
|
||||
panic!(
|
||||
"Python script failed:\nSTDOUT: {}\nSTDERR: {}",
|
||||
String::from_utf8_lossy(&output.stdout),
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
}
|
||||
String::from_utf8_lossy(&output.stdout).trim().to_string()
|
||||
}
|
||||
|
||||
/// Writes `f{i}` (filtered) and `r{i}` (unfiltered twin) for every filter
|
||||
/// setting in `FILTERS` × every case; prints the number of pairs.
|
||||
const GENERATE: &str = r#"
|
||||
import sys
|
||||
import numpy as np, h5py
|
||||
try:
|
||||
import hdf5plugin
|
||||
except ImportError:
|
||||
hdf5plugin = None
|
||||
path = sys.argv[1]
|
||||
FILTERS = eval(sys.argv[2])
|
||||
cases = [
|
||||
('<u1', (1000,), (128,), 'ramp'),
|
||||
('<i2', (37, 53), (10, 16), 'ramp'),
|
||||
('<i4', (2000,), (300,), 'ramp'),
|
||||
('>i4', (37, 53), (8, 8), 'ramp'),
|
||||
('<f4', (5, 6, 7), (2, 3, 4), 'ramp'),
|
||||
('<f8', (1500,), (512,), 'ramp'),
|
||||
('<i8', (33, 17), (33, 17), 'ramp'),
|
||||
('<u2', (4097,), (4097,), 'ramp'),
|
||||
('<f8', (3000,), (1000,), 'noise'),
|
||||
('<u1', (5000,), (5000,), 'noise'),
|
||||
('<i4', (1, 1), (1, 1), 'ramp'),
|
||||
('<f8', (100000,), (40000,), 'ramp'),
|
||||
]
|
||||
rng = np.random.default_rng(7)
|
||||
i = 0
|
||||
with h5py.File(path, 'w') as f:
|
||||
for label, kw in FILTERS:
|
||||
for dt, shape, chunks, kind in cases:
|
||||
n = int(np.prod(shape))
|
||||
if kind == 'noise':
|
||||
raw = rng.integers(0, 256, n * np.dtype(dt).itemsize, dtype=np.uint8)
|
||||
data = raw.view(dt)
|
||||
if np.dtype(dt).kind == 'f':
|
||||
data = np.nan_to_num(data)
|
||||
else:
|
||||
base = (np.arange(n) * 3) % 251 + rng.integers(0, 4, n)
|
||||
data = (base / 7).astype(dt) if np.dtype(dt).kind == 'f' else base.astype(dt)
|
||||
data = data.reshape(shape)
|
||||
f.create_dataset(f'f{i}', data=data, chunks=chunks, **kw)
|
||||
f.create_dataset(f'r{i}', data=data)
|
||||
f[f'f{i}'].attrs['case'] = f'{label} {dt} {shape} chunks={chunks} {kind}'
|
||||
i += 1
|
||||
print(i)
|
||||
"#;
|
||||
|
||||
/// Have h5py write every filter setting in `filters` (a Python list of
|
||||
/// `(label, create_dataset kwargs)`), then check that clawhdf5 reads each
|
||||
/// filtered dataset exactly as its unfiltered twin.
|
||||
fn check_h5py_written(tag: &str, filters: &str) {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join(format!("{tag}.h5"));
|
||||
let n: usize = run_python(GENERATE, &[path.to_str().unwrap(), filters])
|
||||
.parse()
|
||||
.unwrap();
|
||||
assert!(n > 0);
|
||||
let file = File::open(&path).unwrap();
|
||||
for i in 0..n {
|
||||
let filtered = file.dataset(&format!("f{i}")).unwrap();
|
||||
let case = format!("{:?}", filtered.attrs().unwrap().get("case"));
|
||||
let got = filtered
|
||||
.read_selection(&Selection::All)
|
||||
.unwrap_or_else(|e| panic!("{tag} f{i} {case}: {e}"));
|
||||
let want = file
|
||||
.dataset(&format!("r{i}"))
|
||||
.unwrap()
|
||||
.read_selection(&Selection::All)
|
||||
.unwrap();
|
||||
assert!(got == want, "{tag} f{i} {case}: data differs");
|
||||
}
|
||||
}
|
||||
|
||||
/// Values the write-direction tests store: row-major, compressible with some
|
||||
/// variation.
|
||||
fn ramp_i32(n: usize) -> Vec<i32> {
|
||||
(0..n).map(|i| ((i * 3) % 251) as i32 - 60).collect()
|
||||
}
|
||||
fn ramp_f64(n: usize) -> Vec<f64> {
|
||||
(0..n).map(|i| ((i * 3) % 251) as f64 / 7.0).collect()
|
||||
}
|
||||
fn ramp_u8(n: usize) -> Vec<u8> {
|
||||
(0..n).map(|i| ((i * 7) % 256) as u8).collect()
|
||||
}
|
||||
|
||||
/// h5py checks the datasets `write_ours` wrote against the same ramps.
|
||||
const VERIFY: &str = r#"
|
||||
import sys
|
||||
import numpy as np, h5py
|
||||
try:
|
||||
import hdf5plugin
|
||||
except ImportError:
|
||||
pass
|
||||
bad = []
|
||||
with h5py.File(sys.argv[1], 'r') as f:
|
||||
for name in f:
|
||||
ds = f[name]
|
||||
n = ds.size
|
||||
k = np.arange(n)
|
||||
if ds.dtype == np.int32:
|
||||
want = ((k * 3) % 251 - 60).astype(np.int32)
|
||||
elif ds.dtype == np.float64:
|
||||
want = ((k * 3) % 251) / 7.0
|
||||
else:
|
||||
want = ((k * 7) % 256).astype(np.uint8)
|
||||
got = ds[()].reshape(-1)
|
||||
if not np.array_equal(got, want):
|
||||
bad.append(name)
|
||||
# The dataset really is filtered by the plugin under test.
|
||||
if int(sys.argv[2]) not in [int(x) for x in ds._filters.keys() if x.isdigit()] \
|
||||
and sys.argv[3] not in ds._filters:
|
||||
bad.append(name + ':filter-missing:' + repr(ds._filters))
|
||||
print('OK' if not bad else 'BAD ' + ' '.join(bad))
|
||||
"#;
|
||||
|
||||
/// Write i32/f64/u8 datasets (1-D and 2-D, partial edge chunks) with
|
||||
/// `configure` applying the filter, then have h5py read them back.
|
||||
fn check_ours_read_by_h5py(
|
||||
tag: &str,
|
||||
filter_id: u16,
|
||||
filter_name: &str,
|
||||
configure: impl Fn(&mut clawhdf5_format::type_builders::DatasetBuilder),
|
||||
) {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join(format!("{tag}_ours.h5"));
|
||||
let mut fb = clawhdf5::FileBuilder::new();
|
||||
{
|
||||
let ds = fb.create_dataset("i32_1d");
|
||||
ds.with_i32_data(&ramp_i32(10_000)).with_chunks(&[3000]);
|
||||
configure(ds);
|
||||
}
|
||||
{
|
||||
let ds = fb.create_dataset("f64_2d");
|
||||
ds.with_f64_data(&ramp_f64(37 * 53))
|
||||
.with_shape(&[37, 53])
|
||||
.with_chunks(&[10, 16]);
|
||||
configure(ds);
|
||||
}
|
||||
{
|
||||
let ds = fb.create_dataset("u8_1d");
|
||||
ds.with_u8_data(&ramp_u8(5000)).with_chunks(&[777]);
|
||||
configure(ds);
|
||||
}
|
||||
{
|
||||
let ds = fb.create_dataset("f64_big");
|
||||
ds.with_f64_data(&ramp_f64(100_000)).with_chunks(&[40_000]);
|
||||
configure(ds);
|
||||
}
|
||||
fb.write(&path).unwrap();
|
||||
|
||||
// clawhdf5 reads its own output.
|
||||
let file = File::open(&path).unwrap();
|
||||
assert_eq!(
|
||||
file.dataset("i32_1d").unwrap().read_i32().unwrap(),
|
||||
ramp_i32(10_000)
|
||||
);
|
||||
assert_eq!(
|
||||
file.dataset("f64_2d").unwrap().read_f64().unwrap(),
|
||||
ramp_f64(37 * 53)
|
||||
);
|
||||
|
||||
let out = run_python(
|
||||
VERIFY,
|
||||
&[path.to_str().unwrap(), &filter_id.to_string(), filter_name],
|
||||
);
|
||||
assert_eq!(out, "OK", "{tag}: h5py could not read our output");
|
||||
}
|
||||
|
||||
#[cfg(feature = "lzf")]
|
||||
#[test]
|
||||
fn lzf_written_by_h5py_reads_exactly() {
|
||||
if !have_python("h5py") {
|
||||
return;
|
||||
}
|
||||
check_h5py_written(
|
||||
"lzf",
|
||||
r#"[('lzf', dict(compression='lzf')),
|
||||
('lzf+shuffle', dict(compression='lzf', shuffle=True)),
|
||||
('lzf+shuffle+fletcher32', dict(compression='lzf', shuffle=True, fletcher32=True))]"#,
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(feature = "lzf")]
|
||||
#[test]
|
||||
fn lzf_written_by_clawhdf5_reads_in_h5py() {
|
||||
if !have_python("h5py") {
|
||||
return;
|
||||
}
|
||||
check_ours_read_by_h5py("lzf", 32000, "lzf", |ds| {
|
||||
ds.with_lzf();
|
||||
});
|
||||
check_ours_read_by_h5py("lzf_noshuffle", 32000, "lzf", |ds| {
|
||||
ds.with_lzf().without_shuffle();
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user