quantumclaw/clawhdf5
3
1
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 1 Wiki Activity

clawhdf5-format: the writer skips optional filters that fail, as libhdf5 does

Browse Source 
FileBuilder stored every chunk of an LZF or Blosc dataset through the
filter with filter mask 0. libhdf5 counts LZF and Blosc output no smaller
than the chunk as a failure of the optional filter and stores the chunk raw
with the filter's mask bit set. For an LZF chunk whose stream was exactly
the chunk's size, the first libhdf5 rewrite stored raw data at the same
size and kept our stale mask 0 in the index, and h5py could no longer read
the dataset.

precompress_chunks now runs chunks through compress_chunk_masked (as
FileEditor does since f7e2ab1), sequentially and on the parallel path, and
build_chunked_data_from_precompressed records each chunk's real mask in
every index the writer builds: single chunk (layout field), Fixed Array and
Extensible Array filtered elements, and version-2 B-tree type 11 records
(create_datasets_parallel goes through the same path). The writer builds
no version-1 B-tree or implicit index. PrecompressedChunks::chunks gains
the mask. Files whose chunks all compress are byte-identical.

Latent only in the unreleased LZF/Blosc writer (added 2026-09-26); no
tagged release writes either filter.

Tests:
- plugin_filters_interop skipped_optional_filters_are_masked_as_libhdf5_masks_them:
  LZF, shuffle+LZF+fletcher32 and Blosc over random, compressible and
  alternating chunks in every index; masks equal an h5py-written twin's;
  h5py r+ rewrites and extends them; h5py, h5dump and our reader read
  every value. Before: 20 of 24 datasets had masks other than h5py's, and
  with that check disabled h5py failed to read the rewritten datasets
  ("filter returned failure during read").
- plugin_filters_interop files_whose_chunks_all_compress_are_unchanged:
  pins the pre-fix bytes of five all-compressing files.
- chunked_write skipped_lzf_chunks_are_masked_in_every_index (fails before:
  mask 0, want 2).

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
osobh
2026-09-26 14:45:51 -05:00
co-authored by Claude Opus 5.5
parent fe377266e1
commit 0e8522cfad
4 changed files with 613 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/clawhdf5-format/src/chunked_write.rs
+111 -12
View File
@@ -17,7 +17,7 @@ use crate::filter_pipeline::{
FILTER_LZF, FILTER_PCODEC, FILTER_PCODEC_NAME, FILTER_SHUFFLE, FILTER_ZSTD, FilterDescription,
FilterPipeline,
};
use crate::filters::compress_chunk;
use crate::filters::compress_chunk_masked;
/// Round a file offset up to the next cache-line boundary.
///
/// This ensures chunk data starts at an address that is a multiple of the
@@ -489,7 +489,12 @@ pub fn split_into_chunks(
#[cfg(feature = "parallel")]
const PARALLEL_COMPRESS_THRESHOLD: usize = 2;
/// Compress all chunks, using parallel compression when beneficial.
/// Compress all chunks, using parallel compression when beneficial, and
/// return each chunk's stored bytes with its filter mask.
///
/// Chunks run through the pipeline as libhdf5 runs them
/// ([`compress_chunk_masked`]): an optional filter that fails — LZF or Blosc
/// output no smaller than its input — is skipped and its mask bit set.
///
/// With the `parallel` feature and more than [`PARALLEL_COMPRESS_THRESHOLD`]
/// filtered chunks, compression runs across rayon threads; otherwise it is
@@ -499,7 +504,7 @@ fn compress_all_chunks(
chunks: &[(Vec<u64>, Vec<u8>)],
pipeline: &Option<FilterPipeline>,
element_size: u32,
) -> Result<Vec<Vec<u8>>, FormatError> {
) -> Result<Vec<(Vec<u8>, u32)>, FormatError> {
#[cfg(feature = "parallel")]
{
if let Some(pl) = pipeline
@@ -508,7 +513,7 @@ fn compress_all_chunks(
use rayon::prelude::*;
return chunks
.par_iter()
.map(|(_offsets, chunk_bytes)| compress_chunk(chunk_bytes, pl, element_size))
.map(|(_offsets, chunk_bytes)| compress_chunk_masked(chunk_bytes, pl, element_size))
.collect();
}
}
@@ -518,9 +523,9 @@ fn compress_all_chunks(
.iter()
.map(|(_offsets, chunk_bytes)| {
if let Some(pl) = pipeline {
compress_chunk(chunk_bytes, pl, element_size)
compress_chunk_masked(chunk_bytes, pl, element_size)
} else {
Ok(chunk_bytes.clone())
Ok((chunk_bytes.clone(), 0))
}
})
.collect()
@@ -798,8 +803,10 @@ pub fn build_fixed_array_at(
/// writer passes eliminates the double-compression that the two-pass layout
/// algorithm previously performed.
pub struct PrecompressedChunks {
/// Per-chunk: (raw_size_bytes, compressed_bytes).
pub chunks: Vec<(u64, Vec<u8>)>,
/// Per-chunk: (raw_size_bytes, stored_bytes, filter_mask). Bit `i` of
/// the mask is set when filter `i` was skipped (an optional filter that
/// failed); 0 for every chunk of an unfiltered dataset.
pub chunks: Vec<(u64, Vec<u8>, u32)>,
pub has_filters: bool,
pub element_size: usize,
pub shape: Vec<u64>,
@@ -834,7 +841,7 @@ pub fn precompress_chunks(
let chunks = raw_chunks
.into_iter()
.zip(compressed)
.map(|((_offsets, raw_bytes), c)| (raw_bytes.len() as u64, c))
.map(|((_offsets, raw_bytes), (c, mask))| (raw_bytes.len() as u64, c, mask))
.collect();
Ok(PrecompressedChunks {
@@ -867,7 +874,7 @@ pub fn build_chunked_data_from_precompressed(
let mut data_buf = Vec::new();
let mut written_chunks = Vec::with_capacity(num_chunks);
for (raw_size, compressed) in &pre.chunks {
for (raw_size, compressed, filter_mask) in &pre.chunks {
let aligned_offset = align_to_cache_line(data_buf.len());
if aligned_offset > data_buf.len() {
data_buf.resize(aligned_offset, 0u8);
@@ -879,7 +886,7 @@ pub fn build_chunked_data_from_precompressed(
address,
compressed_size,
raw_size: *raw_size,
filter_mask: 0,
filter_mask: *filter_mask,
});
}
@@ -916,7 +923,7 @@ pub fn build_chunked_data_from_precompressed(
} else {
None
};
let filter_mask = if pre.has_filters { Some(0u32) } else { None };
let filter_mask = pre.has_filters.then_some(written_chunks[0].filter_mask);
serialize_v4_single_chunk(
&chunk_dims_u32,
chunk_addr,
@@ -1943,6 +1950,98 @@ mod tests {
bytes_to_f64(&output)
}
/// Every chunk index the writer builds records each chunk's real filter
/// mask: LZF output no smaller than the chunk is skipped (bit 1, behind
/// shuffle) and the chunk stored shuffled only; compressible chunks keep
/// mask 0. The data reads back through both kinds of chunk.
#[cfg(feature = "lzf")]
#[test]
fn skipped_lzf_chunks_are_masked_in_every_index() {
let c = 64usize;
// Chunks alternate: random bytes (LZF cannot shrink them), then 7s.
let mut state = 0x1234_5678_u64;
let data: Vec<f64> = (0..4 * c)
.map(|i| {
if (i / c).is_multiple_of(2) {
state ^= state << 13;
state ^= state >> 7;
state ^= state << 17;
f64::from_bits(state)
} else {
7.0
}
})
.collect();
let raw = f64_to_bytes(&data);
let options = ChunkOptions {
plugin: Some(PluginFilter::Lzf),
..Default::default()
};
let c64 = c as u64;
#[allow(clippy::type_complexity)]
let cases: [(&[u64], &[u64], Option<&[u64]>, u8, &[u32]); 4] = [
(&[c64], &[c64], None, 1, &[2]),
(&[4 * c64], &[c64], None, 3, &[2, 0, 2, 0]),
(&[4 * c64], &[c64], Some(&[u64::MAX]), 4, &[2, 0, 2, 0]),
(
&[2, 2 * c64],
&[1, c64],
Some(&[u64::MAX, u64::MAX]),
5,
&[2, 0, 2, 0],
),
];
let base = 0x1000u64;
for (shape, chunks, maxshape, index_type, want_masks) in cases {
let n: u64 = shape.iter().product();
let raw = &raw[..n as usize * 8];
let result =
build_chunked_data_at_ext(raw, shape, chunks, 8, &options, base, maxshape).unwrap();
let mut file = vec![0u8; base as usize];
file.extend_from_slice(&result.data_bytes);
let layout = DataLayout::parse(&result.layout_message, 8, 8).unwrap();
assert!(
matches!(&layout, DataLayout::Chunked { chunk_index_type, .. }
if *chunk_index_type == Some(index_type)),
"{layout:?}"
);
let dataspace = Dataspace {
space_type: DataspaceType::Simple,
rank: shape.len() as u8,
dimensions: shape.to_vec(),
max_dimensions: maxshape.map(<[u64]>::to_vec),
};
let (mut infos, _) =
crate::chunked_read::list_chunks(&file, &layout, &dataspace, 8, 8, 8).unwrap();
infos.sort_by(|a, b| a.offsets.cmp(&b.offsets));
let masks: Vec<u32> = infos.iter().map(|i| i.filter_mask).collect();
assert_eq!(masks, want_masks, "index type {index_type}");
for info in &infos {
// Skipped chunks are stored at the chunk's size (shuffled).
assert_eq!(
info.chunk_size == (c * 8) as u32,
info.filter_mask != 0,
"{info:?}"
);
}
let pipeline = crate::filter_pipeline::FilterPipeline::parse(
result.pipeline_message.as_ref().unwrap(),
)
.unwrap();
let out = read_chunked_data(
&file,
&layout,
&dataspace,
&make_f64_type(),
Some(&pipeline),
8,
8,
)
.unwrap();
assert_eq!(out, raw, "index type {index_type}");
}
}
#[test]
fn ea_roundtrip_1d_inline_only() {
let values: Vec<f64> = (0..10).map(|i| i as f64).collect();