From 6dfd2390112d132a517388126407a0d3610290ed Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 26 Sep 2026 00:05:26 -0500 Subject: [PATCH] feat(format): bzip2 filter (307), read and write, pure Rust hdf5plugin's BZip2 failed with UnsupportedFilter(307). The new `bzip2` feature decodes the single bzip2 stream H5Zbzip2.c stores, bounded by the chunk size (a truncated stream is an error, not short data), and encodes at block size cd_values[0] (DatasetBuilder::with_bzip2(level)). It uses the bzip2 crate's default backend, libbz2-rs-sys, a pure-Rust port of libbzip2: `cargo tree` shows no cc/cmake, and nothing is compiled from C. Interop: hdf5plugin writes block sizes 9, 1 and 5+shuffle over the 12-case matrix, read byte for byte; ours at 9 (shuffled) and 1 (not) reads back through hdf5plugin. Both fail with the decoder removed. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/clawhdf5-format/Cargo.toml | 5 + crates/clawhdf5-format/src/chunked_write.rs | 18 ++- crates/clawhdf5-format/src/filters.rs | 7 ++ crates/clawhdf5-format/src/filters_bzip2.rs | 112 ++++++++++++++++++ crates/clawhdf5-format/src/lib.rs | 2 + crates/clawhdf5-format/src/type_builders.rs | 7 ++ crates/clawhdf5/Cargo.toml | 1 + .../clawhdf5/tests/plugin_filters_interop.rs | 28 +++++ 8 files changed, 178 insertions(+), 2 deletions(-) create mode 100644 crates/clawhdf5-format/src/filters_bzip2.rs diff --git a/crates/clawhdf5-format/Cargo.toml b/crates/clawhdf5-format/Cargo.toml index f6a9087..007d8e7 100644 --- a/crates/clawhdf5-format/Cargo.toml +++ b/crates/clawhdf5-format/Cargo.toml @@ -25,6 +25,9 @@ pco = { version = "1.0", optional = true } # Pure-Rust Zstandard, for the plugin filters that embed zstd (bitshuffle, # blosc). The `zstd` feature (filter 32015) links libzstd instead. ruzstd = { version = "0.9", optional = true } +# bzip2 with its default backend, libbz2-rs-sys: a pure-Rust port of +# libbzip2 (no C is compiled, despite the -sys name). +bzip2 = { version = "0.6", optional = true } [dev-dependencies] half = { workspace = true } @@ -64,6 +67,8 @@ pcodec = ["dep:pco"] lzf = [] # Bitshuffle (32008), with its LZ4 and Zstandard modes. bitshuffle = ["lz4_flex", "ruzstd"] +# bzip2 (307). +bzip2 = ["dep:bzip2", "std"] [[bench]] name = "parallel_decompress_bench" diff --git a/crates/clawhdf5-format/src/chunked_write.rs b/crates/clawhdf5-format/src/chunked_write.rs index e155c89..e825aed 100644 --- a/crates/clawhdf5-format/src/chunked_write.rs +++ b/crates/clawhdf5-format/src/chunked_write.rs @@ -12,8 +12,9 @@ use crate::chunk_grid::ChunkGrid; use crate::ea_writer; use crate::error::FormatError; use crate::filter_pipeline::{ - FILTER_BITSHUFFLE, FILTER_DEFLATE, FILTER_FLETCHER32, FILTER_LZ4, FILTER_LZF, FILTER_PCODEC, - FILTER_PCODEC_NAME, FILTER_SHUFFLE, FILTER_ZSTD, FilterDescription, FilterPipeline, + FILTER_BITSHUFFLE, FILTER_BZIP2, FILTER_DEFLATE, FILTER_FLETCHER32, FILTER_LZ4, FILTER_LZF, + FILTER_PCODEC, FILTER_PCODEC_NAME, FILTER_SHUFFLE, FILTER_ZSTD, FilterDescription, + FilterPipeline, }; use crate::filters::compress_chunk; /// Round a file offset up to the next cache-line boundary. @@ -70,6 +71,12 @@ pub enum PluginFilter { /// Compression after the transpose. compression: BitshuffleCompression, }, + /// bzip2 (filter 307) at block size `level` (1-9). Needs the `bzip2` + /// feature. + Bzip2 { + /// Block size 1-9 (9 = hdf5plugin's default). + level: u32, + }, } /// What bitshuffle compresses its blocks with. @@ -94,6 +101,7 @@ impl PluginFilter { match self { PluginFilter::Lzf => false, PluginFilter::Bitshuffle { .. } => true, + PluginFilter::Bzip2 { .. } => false, } } @@ -112,6 +120,12 @@ impl PluginFilter { }, // bshuf_h5_set_local: version 0.4, element size, block size, // compression (0 none, 2 LZ4, 3 Zstandard), Zstandard level. + PluginFilter::Bzip2 { level } => FilterDescription { + filter_id: FILTER_BZIP2, + name: Some("bzip2".into()), + flags: 1, + client_data: vec![(*level).clamp(1, 9)], + }, PluginFilter::Bitshuffle { block_size, compression, diff --git a/crates/clawhdf5-format/src/filters.rs b/crates/clawhdf5-format/src/filters.rs index 5a61251..d744cd0 100644 --- a/crates/clawhdf5-format/src/filters.rs +++ b/crates/clawhdf5-format/src/filters.rs @@ -185,6 +185,13 @@ pub(crate) static BUILTIN_FILTERS: &[BuiltinFilter] = &[ decode: |d, c| scaleoffset_decompress(d, c.client_data(), c.max_output), encode: None, }, + #[cfg(feature = "bzip2")] + BuiltinFilter { + id: crate::filter_pipeline::FILTER_BZIP2, + name: "bzip2", + decode: crate::filters_bzip2::bzip2_decode, + encode: Some(crate::filters_bzip2::bzip2_encode), + }, #[cfg(feature = "pcodec")] BuiltinFilter { id: FILTER_PCODEC, diff --git a/crates/clawhdf5-format/src/filters_bzip2.rs b/crates/clawhdf5-format/src/filters_bzip2.rs new file mode 100644 index 0000000..2a78074 --- /dev/null +++ b/crates/clawhdf5-format/src/filters_bzip2.rs @@ -0,0 +1,112 @@ +//! bzip2 (HDF5 filter 307, PyTables' `H5Zbzip2.c`, hdf5plugin's `BZip2`). +//! +//! The chunk is one bzip2 stream; `cd_values[0]` is the block size (1-9, +//! the compression level). Decoded with the `bzip2` crate's default backend, +//! `libbz2-rs-sys`, a pure-Rust port of libbzip2. + +use crate::error::FormatError; +use crate::filter_registry::FilterContext; + +fn err(msg: &str) -> FormatError { + FormatError::DecompressionError(format!("bzip2: {msg}")) +} + +/// Decode a bzip2-filtered chunk, refusing output beyond the chunk size. +pub(crate) fn bzip2_decode(input: &[u8], ctx: &FilterContext<'_>) -> Result, FormatError> { + use bzip2::{Decompress, Status}; + let limit = ctx.output_limit(); + let max_capacity = limit.saturating_add(1); + let hint = if ctx.max_output != 0 { + ctx.max_output + } else { + input.len().saturating_mul(4) + }; + let mut out = Vec::new(); + out.try_reserve_exact(hint.clamp(1, max_capacity)) + .map_err(|_| err("cannot allocate the output buffer"))?; + let mut dec = Decompress::new(false); + loop { + let (in_before, out_before) = (dec.total_in(), dec.total_out()); + let status = dec + .decompress_vec(&input[in_before as usize..], &mut out) + .map_err(|e| err(&e.to_string()))?; + if out.len() > limit { + return Err(err("output exceeds the chunk size")); + } + if status == Status::StreamEnd { + return Ok(out); + } + if out.len() == out.capacity() { + let grow = out.capacity().min(max_capacity - out.capacity()).max(1); + out.try_reserve_exact(grow) + .map_err(|_| err("cannot allocate the output buffer"))?; + } else if dec.total_in() as usize >= input.len() + || (dec.total_in(), dec.total_out()) == (in_before, out_before) + { + return Err(err("truncated stream")); + } + } +} + +/// Encode a chunk as one bzip2 stream at block size `cd_values[0]` +/// (default 9, as hdf5plugin). +pub(crate) fn bzip2_encode(input: &[u8], ctx: &FilterContext<'_>) -> Result, FormatError> { + use bzip2::{Action, Compress, Compression, Status}; + let level = ctx.client_data().first().copied().unwrap_or(9).clamp(1, 9); + let cerr = |m: String| FormatError::CompressionError(format!("bzip2: {m}")); + let mut enc = Compress::new(Compression::new(level), 0); + // bzip2's worst case is about 1% + 600 bytes over the input. + let mut out = Vec::with_capacity(input.len() + input.len() / 100 + 600); + loop { + let consumed = enc.total_in() as usize; + let status = enc + .compress_vec(&input[consumed..], &mut out, Action::Finish) + .map_err(|e| cerr(e.to_string()))?; + if status == Status::StreamEnd { + return Ok(out); + } + if out.len() == out.capacity() { + out.reserve(out.capacity().max(4096)); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::filter_pipeline::{FILTER_BZIP2, FilterDescription}; + + fn desc(level: u32) -> FilterDescription { + FilterDescription { + filter_id: FILTER_BZIP2, + name: None, + flags: 0, + client_data: vec![level], + } + } + + #[test] + fn round_trips_and_bounds() { + let data: Vec = (0..100_000u32) + .flat_map(|i| (i % 777).to_le_bytes()) + .collect(); + for level in [1, 5, 9] { + let f = desc(level); + let ctx = FilterContext { + filter: &f, + element_size: 4, + max_output: data.len(), + }; + let enc = bzip2_encode(&data, &ctx).unwrap(); + assert!(enc.len() < data.len() / 4); + assert_eq!(bzip2_decode(&enc, &ctx).unwrap(), data); + // Truncated, and larger than the chunk: errors, not data. + assert!(bzip2_decode(&enc[..enc.len() / 2], &ctx).is_err()); + let small = FilterContext { + max_output: data.len() - 1, + ..ctx + }; + assert!(bzip2_decode(&enc, &small).is_err()); + } + } +} diff --git a/crates/clawhdf5-format/src/lib.rs b/crates/clawhdf5-format/src/lib.rs index e043567..b27ca09 100644 --- a/crates/clawhdf5-format/src/lib.rs +++ b/crates/clawhdf5-format/src/lib.rs @@ -75,6 +75,8 @@ pub mod filter_registry; pub mod filters; #[cfg(feature = "bitshuffle")] mod filters_bitshuffle; +#[cfg(feature = "bzip2")] +mod filters_bzip2; #[cfg(feature = "lzf")] pub mod filters_lzf; mod filters_szip; diff --git a/crates/clawhdf5-format/src/type_builders.rs b/crates/clawhdf5-format/src/type_builders.rs index d88063c..635d66a 100644 --- a/crates/clawhdf5-format/src/type_builders.rs +++ b/crates/clawhdf5-format/src/type_builders.rs @@ -763,6 +763,13 @@ impl DatasetBuilder { }) } + /// Enable bzip2 (filter 307) at block size `level` (1-9). Implies + /// chunked storage; shuffle is applied first unless + /// `.without_shuffle()`. Requires the `bzip2` cargo feature. + pub fn with_bzip2(&mut self, level: u32) -> &mut Self { + self.with_plugin_filter(crate::chunked_write::PluginFilter::Bzip2 { level }) + } + /// Enable Pcodec lossless numerical compression (private clawhdf5 filter /// ID 480). /// diff --git a/crates/clawhdf5/Cargo.toml b/crates/clawhdf5/Cargo.toml index 52b8ed3..1523920 100644 --- a/crates/clawhdf5/Cargo.toml +++ b/crates/clawhdf5/Cargo.toml @@ -45,6 +45,7 @@ pcodec = ["clawhdf5-format/pcodec"] # compression; it has no dependencies, so it is on by default. lzf = ["clawhdf5-format/lzf"] bitshuffle = ["clawhdf5-format/bitshuffle"] +bzip2 = ["clawhdf5-format/bzip2"] # 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. diff --git a/crates/clawhdf5/tests/plugin_filters_interop.rs b/crates/clawhdf5/tests/plugin_filters_interop.rs index f8e44a1..5cd1188 100644 --- a/crates/clawhdf5/tests/plugin_filters_interop.rs +++ b/crates/clawhdf5/tests/plugin_filters_interop.rs @@ -275,6 +275,34 @@ fn bitshuffle_written_by_clawhdf5_reads_in_hdf5plugin() { } } +#[cfg(feature = "bzip2")] +#[test] +fn bzip2_written_by_hdf5plugin_reads_exactly() { + if !have_python("h5py, hdf5plugin") { + return; + } + check_h5py_written( + "bzip2", + r#"[('bzip2 9', hdf5plugin.BZip2()), + ('bzip2 1', hdf5plugin.BZip2(blocksize=1)), + ('bzip2 5 + shuffle', dict(**hdf5plugin.BZip2(blocksize=5), shuffle=True))]"#, + ); +} + +#[cfg(feature = "bzip2")] +#[test] +fn bzip2_written_by_clawhdf5_reads_in_hdf5plugin() { + if !have_python("h5py, hdf5plugin") { + return; + } + check_ours_read_by_h5py("bzip2", 307, "bzip2", |ds| { + ds.with_bzip2(9); + }); + check_ours_read_by_h5py("bzip2_1", 307, "bzip2", |ds| { + ds.with_bzip2(1).without_shuffle(); + }); +} + #[cfg(feature = "lzf")] #[test] fn lzf_written_by_h5py_reads_exactly() {