diff --git a/crates/clawhdf5-format/Cargo.toml b/crates/clawhdf5-format/Cargo.toml index 007d8e7..75fd233 100644 --- a/crates/clawhdf5-format/Cargo.toml +++ b/crates/clawhdf5-format/Cargo.toml @@ -28,6 +28,7 @@ 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 } +snap = { version = "1", optional = true } [dev-dependencies] half = { workspace = true } @@ -69,6 +70,10 @@ lzf = [] bitshuffle = ["lz4_flex", "ruzstd"] # bzip2 (307). bzip2 = ["dep:bzip2", "std"] +# Blosc 1 (32001) with its BloscLZ, LZ4, Snappy, Zlib and Zstandard codecs. +blosc = ["lz4_flex", "ruzstd", "snap", "deflate", "std"] +# Every plugin filter above. +plugin-filters = ["lzf", "bitshuffle", "bzip2", "blosc"] [[bench]] name = "parallel_decompress_bench" diff --git a/crates/clawhdf5-format/src/chunked_write.rs b/crates/clawhdf5-format/src/chunked_write.rs index e825aed..fef4b4f 100644 --- a/crates/clawhdf5-format/src/chunked_write.rs +++ b/crates/clawhdf5-format/src/chunked_write.rs @@ -12,8 +12,8 @@ use crate::chunk_grid::ChunkGrid; use crate::ea_writer; use crate::error::FormatError; use crate::filter_pipeline::{ - FILTER_BITSHUFFLE, FILTER_BZIP2, FILTER_DEFLATE, FILTER_FLETCHER32, FILTER_LZ4, FILTER_LZF, - FILTER_PCODEC, FILTER_PCODEC_NAME, FILTER_SHUFFLE, FILTER_ZSTD, FilterDescription, + FILTER_BITSHUFFLE, FILTER_BLOSC, 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; @@ -77,6 +77,41 @@ pub enum PluginFilter { /// Block size 1-9 (9 = hdf5plugin's default). level: u32, }, + /// Blosc 1 (filter 32001): `codec` at `level` (0-9; 0 stores), after + /// `shuffle`. Needs the `blosc` feature. + Blosc { + /// The codec inside the Blosc frame. + codec: BloscCodec, + /// Compression level 0-9 (0 stores the data uncompressed). + level: u32, + /// The shuffle Blosc applies first. + shuffle: BloscShuffle, + }, +} + +/// The codec inside a Blosc frame that clawhdf5 can write. (It reads +/// BloscLZ too, but cannot write it.) +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BloscCodec { + /// LZ4. + Lz4, + /// Snappy. + Snappy, + /// Zlib, at the Blosc level. + Zlib, + /// Zstandard (clawhdf5's pure-Rust encoder has one level, about zstd 1). + Zstd, +} + +/// The shuffle Blosc applies before compressing. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BloscShuffle { + /// None. + None, + /// Byte shuffle (Blosc's default). + Byte, + /// Bit shuffle. + Bit, } /// What bitshuffle compresses its blocks with. @@ -102,6 +137,7 @@ impl PluginFilter { PluginFilter::Lzf => false, PluginFilter::Bitshuffle { .. } => true, PluginFilter::Bzip2 { .. } => false, + PluginFilter::Blosc { .. } => true, } } @@ -120,6 +156,35 @@ impl PluginFilter { }, // bshuf_h5_set_local: version 0.4, element size, block size, // compression (0 none, 2 LZ4, 3 Zstandard), Zstandard level. + // hdf5-blosc's blosc_set_local: filter revision 2, Blosc format + // 2, type size, chunk size, then level, shuffle, compressor. + PluginFilter::Blosc { + codec, + level, + shuffle, + } => FilterDescription { + filter_id: FILTER_BLOSC, + name: Some("blosc".into()), + flags: 1, + client_data: vec![ + 2, + 2, + element_size, + chunk_bytes, + (*level).min(9), + match shuffle { + BloscShuffle::None => 0, + BloscShuffle::Byte => 1, + BloscShuffle::Bit => 2, + }, + match codec { + BloscCodec::Lz4 => 1, + BloscCodec::Snappy => 3, + BloscCodec::Zlib => 4, + BloscCodec::Zstd => 5, + }, + ], + }, PluginFilter::Bzip2 { level } => FilterDescription { filter_id: FILTER_BZIP2, name: Some("bzip2".into()), diff --git a/crates/clawhdf5-format/src/filters.rs b/crates/clawhdf5-format/src/filters.rs index d744cd0..aa8847c 100644 --- a/crates/clawhdf5-format/src/filters.rs +++ b/crates/clawhdf5-format/src/filters.rs @@ -206,6 +206,13 @@ pub(crate) static BUILTIN_FILTERS: &[BuiltinFilter] = &[ decode: crate::filters_lzf::lzf_decode, encode: Some(crate::filters_lzf::lzf_encode), }, + #[cfg(feature = "blosc")] + BuiltinFilter { + id: crate::filter_pipeline::FILTER_BLOSC, + name: "blosc", + decode: crate::filters_blosc::blosc_decode, + encode: Some(crate::filters_blosc::blosc_encode), + }, #[cfg(feature = "lz4")] BuiltinFilter { id: FILTER_LZ4, diff --git a/crates/clawhdf5-format/src/filters_bitshuffle.rs b/crates/clawhdf5-format/src/filters_bitshuffle.rs index 9844adc..83bab22 100644 --- a/crates/clawhdf5-format/src/filters_bitshuffle.rs +++ b/crates/clawhdf5-format/src/filters_bitshuffle.rs @@ -218,7 +218,7 @@ pub(crate) fn bitshuffle_decode( } /// Decode Zstandard frames into exactly `dst`, failing if they hold more. -#[cfg(feature = "bitshuffle")] +#[cfg(any(feature = "bitshuffle", feature = "blosc"))] pub(crate) fn zstd_decode_into( decoder: &mut ruzstd::decoding::FrameDecoder, frames: &[u8], @@ -231,7 +231,7 @@ pub(crate) fn zstd_decode_into( /// Compress with ruzstd. It implements one level (roughly zstd's level 1), /// so the requested level only matters to other encoders. -#[cfg(feature = "bitshuffle")] +#[cfg(any(feature = "bitshuffle", feature = "blosc"))] pub(crate) fn zstd_encode(data: &[u8]) -> Vec { ruzstd::encoding::compress_to_vec(data, ruzstd::encoding::CompressionLevel::Fastest) } diff --git a/crates/clawhdf5-format/src/filters_blosc.rs b/crates/clawhdf5-format/src/filters_blosc.rs new file mode 100644 index 0000000..0730c15 --- /dev/null +++ b/crates/clawhdf5-format/src/filters_blosc.rs @@ -0,0 +1,604 @@ +//! Blosc 1 (HDF5 filter 32001, `hdf5-blosc`, hdf5plugin's `Blosc`), in pure +//! Rust: the Blosc 1 frame, its byte shuffle and bit shuffle, and the +//! BloscLZ, LZ4/LZ4HC, Snappy, Zlib and Zstandard codecs inside it. +//! +//! **Frame** (c-blosc 1.x, format version 2). A 16-byte header — version +//! (2), codec format version (1), flags, type size, then little-endian `u32` +//! decoded size, block size and frame size. Flags: bit 0 byte shuffle, bit +//! 1 stored raw ("memcpyed": the data follows the header), bit 2 bit +//! shuffle, bit 4 "do not split", bits 5-7 the codec (0 BloscLZ, 1 LZ4 and +//! LZ4HC, 2 Snappy, 3 Zlib, 4 Zstandard). Unless stored raw, a table of +//! `u32` block offsets follows, one per block of `block size` bytes (the +//! last one may be shorter). A block is one stream, or — when the "do not +//! split" flag is clear, the type size is at most 16, the block holds at +//! least 128 elements, and it is not the short last block — `type size` +//! streams, one per byte plane. Each stream is a `u32` length and the +//! codec's output; a length equal to the stream's decoded size means the +//! bytes are stored raw. The decoded block is then unshuffled (byte shuffle +//! for type size > 1; bit shuffle when the block holds a multiple of 8 +//! elements, the trailing partial element copied as is). +//! +//! **Filter** (`blosc_filter.c`) `cd_values`: `[0]` filter revision, `[1]` +//! Blosc format version, `[2]` type size, `[3]` chunk size in bytes, `[4]` +//! compression level, `[5]` shuffle (0 none, 1 byte, 2 bit), `[6]` +//! compressor (0 blosclz, 1 lz4, 2 lz4hc, 3 snappy, 4 zlib, 5 zstd). The +//! decoder needs only the frame. + +use crate::error::FormatError; +use crate::filter_registry::FilterContext; +use crate::filters_bitshuffle::{bitshuffle_block, bitunshuffle_block}; + +const HEADER: usize = 16; +const FLAG_SHUFFLE: u8 = 0x01; +const FLAG_MEMCPYED: u8 = 0x02; +const FLAG_BITSHUFFLE: u8 = 0x04; +const FLAG_FUTURE: u8 = 0x08; +const FLAG_DONT_SPLIT: u8 = 0x10; +const MAX_SPLITS: usize = 16; +const MIN_BUFFERSIZE: usize = 128; + +fn err(msg: &str) -> FormatError { + FormatError::DecompressionError(format!("blosc: {msg}")) +} + +fn le32(b: &[u8], at: usize) -> Result { + b.get(at..at + 4) + .map(|s| u32::from_le_bytes(s.try_into().unwrap()) as usize) + .ok_or_else(|| err("truncated frame")) +} + +/// The codec inside a Blosc frame (flags bits 5-7). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Codec { + BloscLz, + Lz4, + Snappy, + Zlib, + Zstd, +} + +impl Codec { + fn from_flags(flags: u8) -> Result { + match flags >> 5 { + 0 => Ok(Codec::BloscLz), + 1 => Ok(Codec::Lz4), + 2 => Ok(Codec::Snappy), + 3 => Ok(Codec::Zlib), + 4 => Ok(Codec::Zstd), + other => Err(err(&format!("unknown codec {other}"))), + } + } +} + +/// Decode one codec stream into exactly `dst`. +fn decode_stream( + codec: Codec, + src: &[u8], + dst: &mut [u8], + zstd: &mut Option, +) -> Result<(), FormatError> { + let n = match codec { + Codec::BloscLz => blosclz_decompress(src, dst), + Codec::Lz4 => { + lz4_flex::block::decompress_into(src, dst).map_err(|e| err(&format!("lz4: {e}")))? + } + Codec::Snappy => { + let len = snap::raw::decompress_len(src).map_err(|e| err(&format!("snappy: {e}")))?; + if len != dst.len() { + return Err(err("snappy stream has the wrong size")); + } + snap::raw::Decoder::new() + .decompress(src, dst) + .map_err(|e| err(&format!("snappy: {e}")))? + } + Codec::Zlib => { + let out = crate::filters::inflate_bounded(src, dst.len(), dst.len()) + .map_err(|e| err(&format!("zlib: {e}")))?; + let n = out.len(); + if n == dst.len() { + dst.copy_from_slice(&out); + } + n + } + Codec::Zstd => crate::filters_bitshuffle::zstd_decode_into( + zstd.get_or_insert_with(ruzstd::decoding::FrameDecoder::new), + src, + dst, + )?, + }; + if n != dst.len() { + return Err(err("stream decoded to the wrong size")); + } + Ok(()) +} + +/// Decode a Blosc-filtered chunk: one Blosc 1 frame. +pub(crate) fn blosc_decode(input: &[u8], ctx: &FilterContext<'_>) -> Result, FormatError> { + blosc_decompress(input, ctx.output_limit()) +} + +/// Decompress a Blosc 1 frame, refusing more than `limit` bytes of output. +pub fn blosc_decompress(input: &[u8], limit: usize) -> Result, FormatError> { + if input.len() < HEADER { + return Err(err("truncated header")); + } + let version = input[0]; + let codec_version = input[1]; + let flags = input[2]; + let typesize = input[3] as usize; + let nbytes = le32(input, 4)?; + let blocksize = le32(input, 8)?; + let cbytes = le32(input, 12)?; + if version != 1 && version != 2 { + return Err(err(&format!( + "frame format version {version} is not Blosc 1 (a Blosc 2 chunk?)" + ))); + } + if flags & FLAG_FUTURE != 0 { + return Err(err("unknown header flags")); + } + if nbytes > limit { + return Err(err("decoded size exceeds the chunk size")); + } + if cbytes > input.len() { + return Err(err("frame is longer than the chunk")); + } + let src = &input[..cbytes]; + if nbytes == 0 { + return Ok(Vec::new()); + } + if blocksize == 0 || typesize == 0 { + return Err(err("bad block or type size")); + } + let mut out = vec![0u8; nbytes]; + if flags & FLAG_MEMCPYED != 0 { + if cbytes != nbytes + HEADER { + return Err(err("stored frame has the wrong size")); + } + out.copy_from_slice(&src[HEADER..]); + return Ok(out); + } + let codec = Codec::from_flags(flags)?; + if codec_version != 1 { + return Err(err(&format!( + "unsupported {codec:?} format version {codec_version}" + ))); + } + let nblocks = nbytes.div_ceil(blocksize); + let leftover = nbytes % blocksize; + if nblocks > (cbytes - HEADER) / 4 { + return Err(err("block table is truncated")); + } + let block_len = blocksize.min(nbytes); + let mut tmp = vec![0u8; block_len]; + let mut zstd = None; + let dont_split = flags & FLAG_DONT_SPLIT != 0; + for j in 0..nblocks { + let is_leftover = j == nblocks - 1 && leftover > 0; + let bsize = if is_leftover { leftover } else { blocksize }; + let nsplits = if !dont_split + && typesize <= MAX_SPLITS + && bsize / typesize >= MIN_BUFFERSIZE + && !is_leftover + { + typesize + } else { + 1 + }; + let neblock = bsize / nsplits; + let mut pos = le32(src, HEADER + 4 * j)?; + let tmp = &mut tmp[..bsize]; + for s in 0..nsplits { + if pos + 4 > src.len() { + return Err(err("block offset out of range")); + } + let clen = le32(src, pos)?; + pos += 4; + let stream = src + .get(pos..pos.saturating_add(clen)) + .ok_or_else(|| err("stream runs past the frame"))?; + let dst = &mut tmp[s * neblock..(s + 1) * neblock]; + if clen == neblock { + dst.copy_from_slice(stream); + } else { + decode_stream(codec, stream, dst, &mut zstd)?; + } + pos += clen; + } + // `bsize` is a whole number of splits by construction (`nsplits` > 1 + // only for full blocks, and c-blosc sizes those in whole elements). + if nsplits * neblock != bsize { + return Err(err("block is not a whole number of streams")); + } + let dest = &mut out[j * blocksize..j * blocksize + bsize]; + unshuffle_block(flags, typesize, tmp, dest); + } + Ok(out) +} + +/// Undo the frame's shuffle on one decoded block. +fn unshuffle_block(flags: u8, typesize: usize, src: &[u8], dest: &mut [u8]) { + let bsize = src.len(); + if flags & FLAG_SHUFFLE != 0 && typesize > 1 { + let n = bsize / typesize; + for i in 0..n { + for b in 0..typesize { + dest[i * typesize + b] = src[b * n + i]; + } + } + dest[n * typesize..].copy_from_slice(&src[n * typesize..]); + } else if flags & FLAG_BITSHUFFLE != 0 && bsize >= typesize { + let n = bsize / typesize; + if n.is_multiple_of(8) { + let body = n * typesize; + bitunshuffle_block(&src[..body], &mut dest[..body], n, typesize); + dest[body..].copy_from_slice(&src[body..]); + } else { + dest.copy_from_slice(src); + } + } else { + dest.copy_from_slice(src); + } +} + +/// BloscLZ decompression (c-blosc 1.21 `blosclz_decompress`): returns the +/// number of bytes written, or 0 on malformed input — exactly as the C +/// decoder, including stopping before a match that ends the stream, so a +/// stream libblosc rejects is rejected here too. +/// +/// Instructions: a control byte `ctrl`. Below 32, a literal run of +/// `ctrl + 1` bytes. Otherwise a match: length `(ctrl >> 5) + 2`, extended +/// by following bytes while they are 255 when the top three bits are all +/// set; distance `((ctrl & 31) << 8) + next byte + 1`, or — when that byte +/// is 255 and the high bits are 31 — a 16-bit big-endian distance plus 8192. +/// The first instruction is always a literal. +pub(crate) fn blosclz_decompress(input: &[u8], out: &mut [u8]) -> usize { + const MAX_DISTANCE: usize = 8191; + let limit = input.len(); + if limit == 0 { + return 0; + } + let mut ip = 1usize; + let mut op = 0usize; + let mut ctrl = (input[0] & 31) as usize; + loop { + if ctrl >= 32 { + let mut len = (ctrl >> 5) - 1; + let ofs = (ctrl & 31) << 8; + if len == 6 { + loop { + if ip + 1 >= limit { + return 0; + } + let code = input[ip] as usize; + ip += 1; + len += code; + if code != 255 { + break; + } + } + } else if ip + 1 >= limit { + return 0; + } + let code = input[ip] as usize; + ip += 1; + len += 3; + // The copy source is `distance` bytes back. + let mut distance = ofs + code + 1; + if code == 255 && ofs == 31 << 8 { + if ip + 1 >= limit { + return 0; + } + let far = ((input[ip] as usize) << 8) + input[ip + 1] as usize; + ip += 2; + distance = far + MAX_DISTANCE + 1; + } + if op + len > out.len() { + return 0; + } + if distance > op { + return 0; + } + if ip >= limit { + break; + } + ctrl = input[ip] as usize; + ip += 1; + let start = op - distance; + if distance >= len { + out.copy_within(start..start + len, op); + } else { + for k in 0..len { + out[op + k] = out[start + k]; + } + } + op += len; + } else { + let run = ctrl + 1; + if op + run > out.len() || ip + run > limit { + return 0; + } + out[op..op + run].copy_from_slice(&input[ip..ip + run]); + op += run; + ip += run; + if ip >= limit { + break; + } + ctrl = input[ip] as usize; + ip += 1; + } + } + op +} + +/// The codec our encoder puts inside the frame. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum EncodeCodec { + Lz4, + Snappy, + Zlib, + Zstd, +} + +impl EncodeCodec { + /// From the filter's `cd_values[6]` compressor code. + fn from_cd(code: u32) -> Result { + match code { + 1 | 2 => Ok(EncodeCodec::Lz4), + 3 => Ok(EncodeCodec::Snappy), + 4 => Ok(EncodeCodec::Zlib), + 5 => Ok(EncodeCodec::Zstd), + 0 => Err(FormatError::CompressionError( + "blosc: clawhdf5 cannot write BloscLZ; choose lz4, snappy, zlib or zstd".into(), + )), + other => Err(FormatError::CompressionError(format!( + "blosc: unknown compressor {other}" + ))), + } + } + + fn flags(self) -> u8 { + (match self { + EncodeCodec::Lz4 => 1, + EncodeCodec::Snappy => 2, + EncodeCodec::Zlib => 3, + EncodeCodec::Zstd => 4, + }) << 5 + } + + fn encode(self, data: &[u8], level: u32) -> Result, FormatError> { + match self { + EncodeCodec::Lz4 => Ok(lz4_flex::block::compress(data)), + EncodeCodec::Snappy => snap::raw::Encoder::new() + .compress_vec(data) + .map_err(|e| FormatError::CompressionError(format!("blosc: snappy: {e}"))), + EncodeCodec::Zlib => crate::filters::deflate_bounded(data, level.min(9)) + .map_err(|e| FormatError::CompressionError(format!("blosc: zlib: {e}"))), + EncodeCodec::Zstd => Ok(crate::filters_bitshuffle::zstd_encode(data)), + } + } +} + +/// Block size our encoder uses: at most 256 KiB, a whole number of +/// elements (and, for bit shuffle, of 8-element groups). +fn encode_block_size(nbytes: usize, typesize: usize, bitshuffle: bool) -> usize { + let unit = if bitshuffle { 8 * typesize } else { typesize }; + let target = (256 * 1024).min(nbytes); + if target < unit { + return nbytes.max(1); + } + target / unit * unit +} + +/// Encode a chunk as one Blosc 1 frame. `cd_values` as hdf5-blosc: +/// `[2]` type size, `[4]` level (0 = store), `[5]` shuffle, `[6]` codec. +pub(crate) fn blosc_encode(input: &[u8], ctx: &FilterContext<'_>) -> Result, FormatError> { + let cd = ctx.client_data(); + let cerr = |m: &str| FormatError::CompressionError(format!("blosc: {m}")); + let typesize = match cd.get(2) { + Some(&t) if t != 0 => t as usize, + _ => ctx.element_size.max(1), + }; + // Blosc records the type size in one byte; c-blosc treats larger types + // as bytes. + let typesize = if typesize > 255 { 1 } else { typesize }; + let level = cd.get(4).copied().unwrap_or(5); + let shuffle = cd.get(5).copied().unwrap_or(1); + let codec = EncodeCodec::from_cd(cd.get(6).copied().unwrap_or(1))?; + let nbytes = input.len(); + if nbytes > i32::MAX as usize - HEADER { + return Err(cerr("chunk too large for a Blosc frame")); + } + let mut flags = codec.flags(); + match shuffle { + 0 => {} + 1 => flags |= FLAG_SHUFFLE, + 2 => flags |= FLAG_BITSHUFFLE, + other => return Err(cerr(&format!("unknown shuffle mode {other}"))), + } + let blocksize = encode_block_size(nbytes, typesize, shuffle == 2); + let header = |flags: u8, blocksize: usize, cbytes: usize| { + let mut h = Vec::with_capacity(HEADER); + h.extend_from_slice(&[2, 1, flags, typesize as u8]); + h.extend_from_slice(&(nbytes as u32).to_le_bytes()); + h.extend_from_slice(&(blocksize as u32).to_le_bytes()); + h.extend_from_slice(&(cbytes as u32).to_le_bytes()); + h + }; + let stored = || { + let mut out = header( + FLAG_MEMCPYED | (flags & !(FLAG_SHUFFLE | FLAG_BITSHUFFLE)), + blocksize, + nbytes + HEADER, + ); + out.extend_from_slice(input); + out + }; + if level == 0 || nbytes == 0 { + return Ok(stored()); + } + + let nblocks = nbytes.div_ceil(blocksize); + let leftover = nbytes % blocksize; + let mut body = Vec::with_capacity(nbytes / 2); + let mut starts = Vec::with_capacity(nblocks); + let table_end = HEADER + 4 * nblocks; + let mut shuffled = vec![0u8; blocksize]; + for j in 0..nblocks { + let is_leftover = j == nblocks - 1 && leftover > 0; + let bsize = if is_leftover { leftover } else { blocksize }; + let block = &input[j * blocksize..j * blocksize + bsize]; + let sh = &mut shuffled[..bsize]; + shuffle_block(flags, typesize, block, sh); + starts.push(table_end + body.len()); + let nsplits = if typesize <= MAX_SPLITS + && bsize / typesize >= MIN_BUFFERSIZE + && !is_leftover + && bsize.is_multiple_of(typesize) + { + typesize + } else { + 1 + }; + let neblock = bsize / nsplits; + for s in 0..nsplits { + let part = &sh[s * neblock..(s + 1) * neblock]; + let comp = codec.encode(part, level)?; + if comp.len() < neblock { + body.extend_from_slice(&(comp.len() as u32).to_le_bytes()); + body.extend_from_slice(&comp); + } else { + body.extend_from_slice(&(neblock as u32).to_le_bytes()); + body.extend_from_slice(part); + } + } + if table_end + body.len() >= nbytes + HEADER { + // Incompressible: store instead, as c-blosc does. + return Ok(stored()); + } + } + // A split block must decode as split: the decoder infers splitting from + // the same rule, which requires a whole number of elements per block. + let cbytes = table_end + body.len(); + let mut out = header(flags, blocksize, cbytes); + for s in starts { + out.extend_from_slice(&(s as u32).to_le_bytes()); + } + out.extend_from_slice(&body); + Ok(out) +} + +/// Apply the frame's shuffle to one block (the inverse of +/// [`unshuffle_block`]). +fn shuffle_block(flags: u8, typesize: usize, src: &[u8], dest: &mut [u8]) { + let bsize = src.len(); + if flags & FLAG_SHUFFLE != 0 && typesize > 1 { + let n = bsize / typesize; + for i in 0..n { + for b in 0..typesize { + dest[b * n + i] = src[i * typesize + b]; + } + } + dest[n * typesize..].copy_from_slice(&src[n * typesize..]); + } else if flags & FLAG_BITSHUFFLE != 0 && bsize >= typesize { + let n = bsize / typesize; + if n.is_multiple_of(8) { + let body = n * typesize; + bitshuffle_block(&src[..body], &mut dest[..body], n, typesize); + dest[body..].copy_from_slice(&src[body..]); + } else { + dest.copy_from_slice(src); + } + } else { + dest.copy_from_slice(src); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::filter_pipeline::{FILTER_BLOSC, FilterDescription}; + + /// A blosclz stream: literal "abc", then a 9-byte match 3 back (a run + /// of "abc"), then literal "Z". + #[test] + fn blosclz_decodes_literals_and_overlapping_matches() { + // Match: length (ctrl >> 5) + 2 = 8, distance ofs + code + 1 = 3. + let stream = [2, b'a', b'b', b'c', (6 << 5), 2, 0, b'Z']; + let mut out = [0u8; 12]; + assert_eq!(blosclz_decompress(&stream, &mut out), 12); + assert_eq!(&out, b"abcabcabcabZ"); + // A stream cut inside a match is malformed. + let mut out = [0u8; 11]; + assert_eq!(blosclz_decompress(&stream[..6], &mut out), 0); + // A match before the start of the output is malformed. + assert_eq!(blosclz_decompress(&[0, b'a', 32, 5, 0, b'x'], &mut out), 0); + } + + fn desc(cd: Vec) -> FilterDescription { + FilterDescription { + filter_id: FILTER_BLOSC, + name: None, + flags: 0, + client_data: cd, + } + } + + #[test] + fn frame_round_trips_every_codec_and_shuffle() { + for ts in [1usize, 2, 4, 8, 3, 32] { + for n in [0usize, 5, 100, 1000, 70_000, 300_001] { + if n * ts > 1 << 20 && ts > 1 { + continue; + } + let data: Vec = (0..n * ts) + .map(|i| ((i / ts) % 200) as u8 ^ (i % ts) as u8) + .collect(); + for codec in [1u32, 3, 4, 5] { + for shuffle in [0u32, 1, 2] { + for level in [0u32, 5] { + let f = desc(vec![2, 2, ts as u32, 0, level, shuffle, codec]); + let ctx = FilterContext { + filter: &f, + element_size: ts, + max_output: data.len(), + }; + let enc = blosc_encode(&data, &ctx).unwrap(); + let dec = blosc_decode(&enc, &ctx).unwrap_or_else(|e| { + panic!("ts={ts} n={n} codec={codec} shuffle={shuffle}: {e}") + }); + assert!( + dec == data, + "ts={ts} n={n} codec={codec} shuffle={shuffle} level={level}" + ); + } + } + } + } + } + } + + #[test] + fn rejects_bad_frames() { + let data = vec![9u8; 50_000]; + let f = desc(vec![2, 2, 4, 0, 5, 1, 1]); + let ctx = FilterContext { + filter: &f, + element_size: 4, + max_output: data.len(), + }; + let enc = blosc_encode(&data, &ctx).unwrap(); + assert!(blosc_decode(&enc[..enc.len() - 3], &ctx).is_err()); + let small = FilterContext { + max_output: 49_999, + ..ctx + }; + assert!(blosc_decode(&enc, &small).is_err()); + let mut v3 = enc.clone(); + v3[0] = 3; + assert!(blosc_decode(&v3, &ctx).is_err()); + let f0 = desc(vec![2, 2, 4, 0, 5, 1, 0]); + let ctx0 = FilterContext { filter: &f0, ..ctx }; + assert!(blosc_encode(&data, &ctx0).is_err()); + } +} diff --git a/crates/clawhdf5-format/src/lib.rs b/crates/clawhdf5-format/src/lib.rs index b27ca09..d9bba65 100644 --- a/crates/clawhdf5-format/src/lib.rs +++ b/crates/clawhdf5-format/src/lib.rs @@ -73,8 +73,10 @@ pub mod fill_value; pub mod filter_pipeline; pub mod filter_registry; pub mod filters; -#[cfg(feature = "bitshuffle")] +#[cfg(any(feature = "bitshuffle", feature = "blosc"))] mod filters_bitshuffle; +#[cfg(feature = "blosc")] +pub mod filters_blosc; #[cfg(feature = "bzip2")] mod filters_bzip2; #[cfg(feature = "lzf")] diff --git a/crates/clawhdf5-format/src/type_builders.rs b/crates/clawhdf5-format/src/type_builders.rs index 635d66a..f055a88 100644 --- a/crates/clawhdf5-format/src/type_builders.rs +++ b/crates/clawhdf5-format/src/type_builders.rs @@ -770,6 +770,22 @@ impl DatasetBuilder { self.with_plugin_filter(crate::chunked_write::PluginFilter::Bzip2 { level }) } + /// Enable Blosc (filter 32001) with `codec` at `level` (0-9) after + /// `shuffle`. Implies chunked storage; no extra HDF5 shuffle is added. + /// Requires the `blosc` cargo feature. + pub fn with_blosc( + &mut self, + codec: crate::chunked_write::BloscCodec, + level: u32, + shuffle: crate::chunked_write::BloscShuffle, + ) -> &mut Self { + self.with_plugin_filter(crate::chunked_write::PluginFilter::Blosc { + codec, + level, + shuffle, + }) + } + /// Enable Pcodec lossless numerical compression (private clawhdf5 filter /// ID 480). /// diff --git a/crates/clawhdf5/Cargo.toml b/crates/clawhdf5/Cargo.toml index 1523920..a6401fe 100644 --- a/crates/clawhdf5/Cargo.toml +++ b/crates/clawhdf5/Cargo.toml @@ -46,6 +46,9 @@ pcodec = ["clawhdf5-format/pcodec"] lzf = ["clawhdf5-format/lzf"] bitshuffle = ["clawhdf5-format/bitshuffle"] bzip2 = ["clawhdf5-format/bzip2"] +blosc = ["clawhdf5-format/blosc"] +# Every plugin filter. +plugin-filters = ["lzf", "bitshuffle", "bzip2", "blosc"] # 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 5cd1188..fc44e92 100644 --- a/crates/clawhdf5/tests/plugin_filters_interop.rs +++ b/crates/clawhdf5/tests/plugin_filters_interop.rs @@ -76,7 +76,7 @@ try: except ImportError: hdf5plugin = None path = sys.argv[1] -FILTERS = eval(sys.argv[2]) +FILTERS = eval('(' + sys.argv[2] + ')') cases = [ ('