Chunked reads beat an h5py process pool; unlimited writer B-trees; Blosc2; 599/697 conformance #16

Merged
osobh merged 48 commits from feat/p2b-scale into main 2026-09-26 17:42:16 +00:00
2 changed files with 59 additions and 13 deletions
Showing only changes of commit e05530a805 - Show all commits
+27 -12
View File
@@ -209,7 +209,10 @@ fn read_header(src: &[u8]) -> Result<ChunkHeader, FormatError> {
if flags2 & VL_BLOCKS != 0 {
return Err(err("variable-length blocks are not supported"));
}
if nbytes > 0 && blocksize > nbytes {
// c-blosc2 clamps only a non-empty chunk's block size; clamping an
// empty one too keeps its scratch blocks from being sized by the
// header (up to 512 MiB each).
if blocksize > nbytes {
blocksize = nbytes;
}
h.blocksize = blocksize;
@@ -234,9 +237,6 @@ pub fn blosc2_decompress_chunk(src: &[u8], limit: usize) -> Result<Vec<u8>, Form
if memcpyed && h.cbytes != h.nbytes + h.overhead {
return Err(err("stored chunk has the wrong size"));
}
if h.nbytes == 0 && h.cbytes == h.overhead && h.special == 0 {
return Ok(Vec::new());
}
let nbytes = h.nbytes;
let mut out = vec![0u8; nbytes];
if h.special != 0 {
@@ -288,6 +288,9 @@ pub fn blosc2_decompress_chunk(src: &[u8], limit: usize) -> Result<Vec<u8>, Form
return Err(err(&format!("filter {f} is not supported")));
}
}
if nbytes == 0 {
return Ok(out);
}
let blocksize = h.blocksize;
let nblocks = nbytes.div_ceil(blocksize);
let leftover = nbytes % blocksize;
@@ -524,7 +527,11 @@ fn parse_frame(buf: &[u8], limit: usize) -> Result<Frame<'_>, FormatError> {
} else {
blosc2_decompress_chunk(off_src, off_len)?
};
let expected = if chunksize > 0 { off_len } else { offsets.len() };
let expected = if chunksize > 0 {
off_len
} else {
offsets.len()
};
if !offsets.len().is_multiple_of(8) || offsets.len() != expected {
return Err(err("offsets chunk does not match the number of chunks"));
}
@@ -548,15 +555,18 @@ impl Frame<'_> {
}
let raw: [u8; 8] = self.offsets[8 * n..8 * n + 8].try_into().unwrap();
let offset = i64::from_le_bytes(raw);
// Every chunk but the last holds `chunksize` bytes.
let size = if self.chunksize == 0 {
None
} else if n == self.nchunks - 1 && !self.nbytes.is_multiple_of(self.chunksize) {
Some(self.nbytes % self.chunksize)
} else {
Some(self.chunksize)
};
if offset < 0 {
// A special chunk, recorded in the offset's top byte.
if self.chunksize == 0 {
let Some(size) = size else {
return Err(err("special chunk in a frame without a chunk size"));
}
let size = if n == self.nchunks - 1 && !self.nbytes.is_multiple_of(self.chunksize) {
self.nbytes % self.chunksize
} else {
self.chunksize
};
if size > limit {
return Err(err("decoded size exceeds the limit"));
@@ -586,7 +596,12 @@ impl Frame<'_> {
.and_then(|o| self.header_len.checked_add(o))
.filter(|&s| s < end && end - s >= MIN_HEADER)
.ok_or_else(|| err("chunk offset out of range"))?;
blosc2_decompress_chunk(&self.buf[start..end], limit)
let data =
blosc2_decompress_chunk(&self.buf[start..end], limit.min(size.unwrap_or(limit)))?;
if size.is_some_and(|s| s != data.len()) {
return Err(err("chunk size does not match the frame's chunk size"));
}
Ok(data)
}
/// The content of metalayer `name`, if the frame has it.
@@ -83,7 +83,13 @@ fn chunk_header(ts: u8, nbytes: i32, blocksize: i32, cbytes: i32, special: u8) -
/// A chunk of `nbytes` bytes that repeats one value (special type 3).
fn repeated(value: &[u8], nbytes: i32, blocksize: i32) -> Vec<u8> {
let mut c = chunk_header(value.len() as u8, nbytes, blocksize, 32 + value.len() as i32, 3);
let mut c = chunk_header(
value.len() as u8,
nbytes,
blocksize,
32 + value.len() as i32,
3,
);
c.extend_from_slice(value);
c
}
@@ -189,3 +195,28 @@ fn small_frames_still_decode() {
assert_eq!(blosc2_decompress(&f, 64).unwrap(), vec![0; 64]);
let _ = blosc2_decompress_chunk;
}
/// A chunk that decodes to nothing kept its declared block size (up to
/// 512 MiB) and allocated two scratch blocks of it: about 1 GiB for a
/// 20-byte chunk.
#[test]
fn empty_chunk_does_not_allocate_its_block_size() {
let _g = lock();
let mut c = vec![5u8, 1, 0x01, 1];
for v in [0i32, 0x1FFF_F000, 20] {
c.extend_from_slice(&v.to_le_bytes());
}
c.resize(20, 0);
let (r, peak) = peak_during(|| blosc2_decompress_chunk(&c, 1 << 20));
assert_eq!(r.map(|v| v.len()).unwrap_or(0), 0);
assert!(
peak <= bound(0, &c),
"peak {peak} bytes for a 20-byte chunk"
);
// Inside a frame for a non-empty HDF5 chunk it is an error, not data.
let offsets = repeated(&0i64.to_le_bytes(), 8, 8);
let f = frame(None, 64, 4, 64, &c, &offsets);
let (r, peak) = peak_during(|| blosc2_decompress(&f, 64));
assert!(r.is_err(), "decoded {:?}", r.map(|v| v.len()));
assert!(peak <= bound(64, &f), "in a frame: peak {peak} bytes");
}