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

Merge branch 'feat/p3-in-place-modify' into feat/p3-range-zfp-edit

Browse Source 
# Conflicts:
#	CHANGELOG.md
#	crates/clawhdf5-py/src/lib.rs
#	crates/clawhdf5/src/error.rs
This commit is contained in:
osobh
2026-09-26 14:52:55 -05:00
parent c233fbca6e 0e8522cfad
commit 8fadb9f424
21 changed files with 5919 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/clawhdf5-format/src/chunked_write.rs
+111 -12
View File
@@ -18,7 +18,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
@@ -490,7 +490,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
@@ -500,7 +505,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
@@ -509,7 +514,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();
}
}
@@ -519,9 +524,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()
@@ -799,8 +804,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>,
@@ -835,7 +842,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 {
@@ -868,7 +875,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);
@@ -880,7 +887,7 @@ pub fn build_chunked_data_from_precompressed(
address,
compressed_size,
raw_size: *raw_size,
filter_mask: 0,
filter_mask: *filter_mask,
});
}
@@ -917,7 +924,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,
@@ -1944,6 +1951,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();