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) <[email protected]>

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"

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_BITSHUFFLE, FILTER_BZIP2, FILTER_DEFLATE, FILTER_FLETCHER32, FILTER_LZ4, FILTER_LZF,
-    FILTER_PCODEC_NAME, FILTER_SHUFFLE, FILTER_ZSTD, FilterDescription, FilterPipeline,
+    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,

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,

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<Vec<u8>, 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<Vec<u8>, 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<u8> = (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());
+        }
+    }
+}

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;

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).
     ///

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.

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() {