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

Merge branch 'fix/p0-writer-meta' into fix/phase0-correctness

Browse Source
This commit is contained in:
osobh
2026-09-25 21:20:59 -05:00
parent e1115bc92a 7c1968a34a
commit c5302e587e
15 changed files with 1457 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/clawhdf5-format/src/file_writer.rs
+181 -56
View File
@@ -4,7 +4,7 @@
//! link messages, contiguous datasets, inline and dense attributes.
#[cfg(not(feature = "std"))]
use alloc::{string::String, string::ToString, vec, vec::Vec};
use alloc::{format, string::String, string::ToString, vec, vec::Vec};
use crate::attribute::AttributeMessage;
use crate::chunked_write::{
@@ -19,7 +19,7 @@ use crate::metadata_index::{DatasetMetadata, MetadataBlock, MetadataIndex};
use crate::object_header_writer::ObjectHeaderWriter;
use crate::superblock::Superblock;
use crate::type_builders::{
DatasetBuilder, FillTime, FinishedGroup, GroupBuilder, build_attr_message,
DatasetBuilder, FinishedGroup, GroupBuilder, build_attr_message, fill_value_message,
};
// Re-export public types that moved to type_builders for API compatibility.
@@ -33,6 +33,49 @@ pub(crate) const OFFSET_SIZE: u8 = 8;
pub(crate) const LENGTH_SIZE: u8 = 8;
const SUPERBLOCK_SIZE: usize = 48;
/// Largest raw data a compact dataset can hold: the layout message (version,
/// class, 2-byte size, data) must fit an object header message, whose size
/// field is 2 bytes. Bigger "compact" requests fall back to contiguous storage.
const MAX_COMPACT_DATA_SIZE: usize = crate::object_header_writer::MAX_MESSAGE_SIZE - 4;
/// libhdf5's bounds on a file space page size (`H5F_FILE_SPACE_PAGE_SIZE_MIN`
/// and `_MAX`).
const MIN_FILE_SPACE_PAGE_SIZE: u32 = 512;
const MAX_FILE_SPACE_PAGE_SIZE: u32 = 1024 * 1024 * 1024;
/// Superblock extension object header for a file using the paged file-space
/// strategy: a single File Space Info message (0x0017), as libhdf5 writes it
/// for `fs_strategy="page"` without persisted free space.
fn build_paged_superblock_extension(page_size: u32) -> Result<Vec<u8>, FormatError> {
let mut fsinfo = Vec::new();
fsinfo.push(1); // version
fsinfo.push(1); // strategy: H5F_FSPACE_STRATEGY_PAGE
fsinfo.push(0); // persisting free space: no
write_length(&mut fsinfo, 1, LENGTH_SIZE); // free-space section threshold
write_length(&mut fsinfo, u64::from(page_size), LENGTH_SIZE);
fsinfo.extend_from_slice(&0u16.to_le_bytes()); // page end metadata threshold
write_undef_offset(&mut fsinfo, OFFSET_SIZE); // EOA before free-space info
let mut w = ObjectHeaderWriter::new();
// Flags as libhdf5 sets them: bit 2 (never share) and bit 4 (mark if
// unknown). Not constant: libhdf5 rewrites the message when it closes a
// file it opened for writing.
w.add_message_with_flags(MessageType::Unknown(0x0017), fsinfo, 0x14);
w.serialize()
}
/// A group or dataset name must be one path component: not empty, not ".",
/// and without '/'. `FileWriter` writes a root group plus one level of
/// groups, and cannot create intermediate groups for a path.
fn check_link_name(name: &str) -> Result<(), FormatError> {
if name.is_empty() || name == "." || name.contains('/') {
return Err(FormatError::SerializationError(format!(
"invalid object name {name:?}: names must be a single path component \
(FileWriter does not create nested groups)"
)));
}
Ok(())
}
/// Threshold for switching from compact (inline) to dense attribute storage.
const DENSE_ATTR_THRESHOLD: usize = 8;
@@ -50,12 +93,12 @@ pub(crate) fn build_chunked_dataset_oh(
pipeline_message: Option<&[u8]>,
attrs: &[AttributeMessage],
dense_blob: Option<&DenseAttrBlob>,
fill_time: FillTime,
) -> Vec<u8> {
fill_message: &[u8],
) -> Result<Vec<u8>, FormatError> {
let mut w = ObjectHeaderWriter::new();
w.add_message_with_flags(MessageType::Datatype, dt.serialize(), 0x01);
w.add_message(MessageType::Dataspace, ds.serialize(LENGTH_SIZE));
w.add_message_with_flags(MessageType::FillValue, vec![3, fill_time.to_byte()], 0x01);
w.add_message_with_flags(MessageType::FillValue, fill_message.to_vec(), 0x01);
w.add_message(MessageType::DataLayout, layout_message.to_vec());
if let Some(pm) = pipeline_message {
w.add_message(MessageType::FilterPipeline, pm.to_vec());
@@ -77,12 +120,12 @@ pub(crate) fn build_dataset_oh(
data_size: u64,
attrs: &[AttributeMessage],
dense_blob: Option<&DenseAttrBlob>,
fill_time: FillTime,
) -> Vec<u8> {
fill_message: &[u8],
) -> Result<Vec<u8>, FormatError> {
let mut w = ObjectHeaderWriter::new();
w.add_message_with_flags(MessageType::Datatype, dt.serialize(), 0x01);
w.add_message(MessageType::Dataspace, ds.serialize(LENGTH_SIZE));
w.add_message_with_flags(MessageType::FillValue, vec![3, fill_time.to_byte()], 0x01);
w.add_message_with_flags(MessageType::FillValue, fill_message.to_vec(), 0x01);
let mut dl = Vec::new();
dl.push(4); // version
dl.push(1); // class = contiguous
@@ -112,12 +155,12 @@ pub(crate) fn build_compact_dataset_oh(
data: &[u8],
attrs: &[AttributeMessage],
dense_blob: Option<&DenseAttrBlob>,
fill_time: FillTime,
) -> Vec<u8> {
fill_message: &[u8],
) -> Result<Vec<u8>, FormatError> {
let mut w = ObjectHeaderWriter::new();
w.add_message_with_flags(MessageType::Datatype, dt.serialize(), 0x01);
w.add_message(MessageType::Dataspace, ds.serialize(LENGTH_SIZE));
w.add_message_with_flags(MessageType::FillValue, vec![3, fill_time.to_byte()], 0x01);
w.add_message_with_flags(MessageType::FillValue, fill_message.to_vec(), 0x01);
// Compact layout message: version=4, class=0, u16 size, inline data
let mut dl = Vec::new();
dl.push(4); // version
@@ -140,7 +183,7 @@ pub(crate) fn build_group_oh(
dense_link_info: Option<&[u8]>,
attrs: &[AttributeMessage],
dense_blob: Option<&DenseAttrBlob>,
) -> Vec<u8> {
) -> Result<Vec<u8>, FormatError> {
let mut w = ObjectHeaderWriter::new();
if let Some(li) = dense_link_info {
// Dense link storage: a LinkInfo pointing at the fractal heap + name
@@ -902,12 +945,12 @@ pub(crate) fn build_vds_dataset_oh(
global_heap_addr: u64,
attrs: &[AttributeMessage],
dense_blob: Option<&DenseAttrBlob>,
fill_time: FillTime,
) -> Vec<u8> {
fill_message: &[u8],
) -> Result<Vec<u8>, FormatError> {
let mut w = ObjectHeaderWriter::new();
w.add_message_with_flags(MessageType::Datatype, dt.serialize(), 0x01);
w.add_message(MessageType::Dataspace, ds.serialize(LENGTH_SIZE));
w.add_message_with_flags(MessageType::FillValue, vec![3, fill_time.to_byte()], 0x01);
w.add_message_with_flags(MessageType::FillValue, fill_message.to_vec(), 0x01);
// VDS layout message: version=4, class=3, global_heap_address(8), global_heap_index=1(4)
let mut dl = Vec::new();
dl.push(4u8); // version
@@ -956,7 +999,9 @@ pub struct FileWriter {
alignment_threshold: usize,
/// Global alignment boundary in bytes (0 = disabled).
alignment_bytes: usize,
/// Page size for page-buffer mode. When set, a v4 superblock is written.
/// File space page size. When set, the file uses libhdf5's paged
/// file-space strategy (a File Space Info message in the superblock
/// extension).
page_size: Option<u32>,
}
@@ -988,9 +1033,16 @@ impl FileWriter {
self
}
/// Enable page-buffer mode with the given page size. Writing this causes
/// the file to be written with a v4 superblock (page_size field) instead
/// of the default v3.
/// Write the file with libhdf5's *paged* file-space strategy and the given
/// page size, as `H5Pset_file_space_strategy(H5F_FSPACE_STRATEGY_PAGE)` +
/// `H5Pset_file_space_page_size` (h5py: `fs_strategy="page"`,
/// `fs_page_size=...`) do: a v3 superblock with an extension holding a
/// File Space Info message, and the file padded to a whole number of
/// pages. Readers with a page buffer can then fetch metadata page by page.
///
/// `page_size` must be between 512 bytes and 1 GiB (libhdf5's limits);
/// [`Self::finish`] fails otherwise. This used to write a "version 4"
/// superblock, which does not exist and no HDF5 library can open.
pub fn with_page_size(&mut self, page_size: u32) -> &mut Self {
self.page_size = Some(page_size);
self
@@ -1015,6 +1067,14 @@ impl FileWriter {
pub fn finish(self) -> Result<Vec<u8>, FormatError> {
let page_size = self.page_size;
if let Some(ps) = page_size
&& !(MIN_FILE_SPACE_PAGE_SIZE..=MAX_FILE_SPACE_PAGE_SIZE).contains(&ps)
{
return Err(FormatError::SerializationError(format!(
"file space page size {ps} is outside libhdf5's \
{MIN_FILE_SPACE_PAGE_SIZE}..={MAX_FILE_SPACE_PAGE_SIZE} bytes"
)));
}
struct DsFlat {
name: String,
dt: Datatype,
@@ -1023,7 +1083,8 @@ impl FileWriter {
attrs: Vec<AttributeMessage>,
chunk_options: ChunkOptions,
maxshape: Option<Vec<u64>>,
fill_time: FillTime,
/// Serialized Fill Value message.
fill_message: Vec<u8>,
compact: bool,
alignment: usize,
/// VDS source mappings (set for Virtual datasets).
@@ -1073,6 +1134,7 @@ impl FileWriter {
};
attrs.extend(p.build_attrs(&raw));
}
let fill_message = fill_value_message(db.fill_time, db.fill_value.as_deref(), &dt)?;
Ok(DsFlat {
name: db.name,
dt,
@@ -1081,13 +1143,26 @@ impl FileWriter {
attrs,
chunk_options: db.chunk_options,
maxshape: db.maxshape,
fill_time: db.fill_time,
fill_message,
compact: db.compact,
alignment: db.alignment,
virtual_sources: db.virtual_sources,
})
};
// Every name becomes a single link in its parent group. The writer
// has no nested groups, so a path like "a/b" would be stored as one
// link literally named "a/b" — which no HDF5 reader can resolve.
let root_names = self.root_datasets.iter().map(|d| d.name.as_str());
let group_names = self.groups.iter().flat_map(|g| {
core::iter::once(g.name.as_str())
.chain(g.datasets.iter().map(|d| d.name.as_str()))
.chain(g.external_links.iter().map(|l| l.0.as_str()))
});
for name in root_names.chain(group_names) {
check_link_name(name)?;
}
let mut all_ds: Vec<DsFlat> = Vec::new();
let mut groups: Vec<GrpFlat> = Vec::new();
let mut root_ds_indices: Vec<usize> = Vec::new();
@@ -1120,6 +1195,17 @@ impl FileWriter {
root_attrs.push(build_attr_message(n, v));
}
// Every datatype must have an on-disk encoding before anything is laid
// out: `Datatype::serialize` itself cannot report a failure.
let group_attrs = groups.iter().flat_map(|g| &g.attrs);
let ds_attrs = all_ds.iter().flat_map(|d| &d.attrs);
for a in root_attrs.iter().chain(group_attrs).chain(ds_attrs) {
a.datatype.check_encodable()?;
}
for d in &all_ds {
d.dt.check_encodable()?;
}
let is_vds: Vec<bool> = all_ds.iter().map(|d| d.virtual_sources.is_some()).collect();
let is_chunked: Vec<bool> = all_ds
.iter()
@@ -1135,7 +1221,9 @@ impl FileWriter {
let is_compact: Vec<bool> = all_ds
.iter()
.enumerate()
.map(|(i, d)| !is_vds[i] && !is_chunked[i] && d.compact && d.raw.len() <= 65535)
.map(|(i, d)| {
!is_vds[i] && !is_chunked[i] && d.compact && d.raw.len() <= MAX_COMPACT_DATA_SIZE
})
.collect();
let root_dense = root_attrs.len() > DENSE_ATTR_THRESHOLD;
let group_dense: Vec<bool> = groups
@@ -1174,9 +1262,9 @@ impl FileWriter {
}
let attr_blob = group_dense[gi].then(|| build_dense_attrs(&g.attrs, 0));
let dl = group_links_dense[gi].then_some(dummy_link_info.as_slice());
build_group_oh(&dummy_links, dl, &g.attrs, attr_blob.as_ref()).len()
build_group_oh(&dummy_links, dl, &g.attrs, attr_blob.as_ref()).map(|oh| oh.len())
})
.collect();
.collect::<Result<_, _>>()?;
let root_dummy_links: Vec<LinkMessage> = {
let mut links = Vec::new();
@@ -1191,7 +1279,7 @@ impl FileWriter {
let root_oh_size = {
let attr_blob = root_dense.then(|| build_dense_attrs(&root_attrs, 0));
let dl = root_links_dense.then_some(dummy_link_info.as_slice());
build_group_oh(&root_dummy_links, dl, &root_attrs, attr_blob.as_ref()).len()
build_group_oh(&root_dummy_links, dl, &root_attrs, attr_blob.as_ref())?.len()
};
struct DataBlob {
@@ -1219,8 +1307,8 @@ impl FileWriter {
0, // dummy address
&d.attrs,
dense_blob.as_ref(),
d.fill_time,
);
&d.fill_message,
)?;
// Global heap blob size is address-independent; compute it now
// so pass 2 can place it correctly.
let vds_mappings = d.virtual_sources.as_deref().unwrap_or(&[]);
@@ -1263,8 +1351,8 @@ impl FileWriter {
result.pipeline_message.as_deref(),
&d.attrs,
dense_blob.as_ref(),
d.fill_time,
);
&d.fill_message,
)?;
dummy_blobs.push(DataBlob {
data: result.data_bytes,
oh_bytes: oh,
@@ -1282,8 +1370,8 @@ impl FileWriter {
&d.raw,
&d.attrs,
dense_blob.as_ref(),
d.fill_time,
);
&d.fill_message,
)?;
dummy_blobs.push(DataBlob {
data: vec![],
oh_bytes: oh,
@@ -1302,8 +1390,8 @@ impl FileWriter {
d.raw.len() as u64,
&d.attrs,
dense_blob.as_ref(),
d.fill_time,
);
&d.fill_message,
)?;
dummy_blobs.push(DataBlob {
data: d.raw.clone(),
oh_bytes: oh,
@@ -1315,12 +1403,12 @@ impl FileWriter {
let actual_ds_oh_sizes: Vec<usize> = dummy_blobs.iter().map(|b| b.oh_bytes.len()).collect();
// Pass 2: compute real addresses
// v4 superblocks add a 4-byte page_size field before the checksum.
let superblock_size = if page_size.is_some() {
SUPERBLOCK_SIZE + 4
} else {
SUPERBLOCK_SIZE
};
// A paged file carries its File Space Info in a superblock extension
// object header, placed right after the superblock.
let sb_ext = page_size
.map(build_paged_superblock_extension)
.transpose()?;
let superblock_size = SUPERBLOCK_SIZE + sb_ext.as_ref().map_or(0, Vec::len);
let root_group_addr = superblock_size as u64;
let mut cursor2 = superblock_size + root_oh_size;
@@ -1411,8 +1499,8 @@ impl FileWriter {
heap_addr,
&d.attrs,
ds_dense_blobs[i].as_ref(),
d.fill_time,
);
&d.fill_message,
)?;
ds_blobs2.push(DataBlob {
data: gcol_bytes.clone(),
oh_bytes: oh,
@@ -1438,8 +1526,8 @@ impl FileWriter {
result.pipeline_message.as_deref(),
&d.attrs,
ds_dense_blobs[i].as_ref(),
d.fill_time,
);
&d.fill_message,
)?;
ds_blobs2.push(DataBlob {
data: result.data_bytes,
oh_bytes: oh,
@@ -1453,8 +1541,8 @@ impl FileWriter {
&d.raw,
&d.attrs,
ds_dense_blobs[i].as_ref(),
d.fill_time,
);
&d.fill_message,
)?;
ds_blobs2.push(DataBlob {
data: vec![],
oh_bytes: oh,
@@ -1478,8 +1566,8 @@ impl FileWriter {
d.raw.len() as u64,
&d.attrs,
ds_dense_blobs[i].as_ref(),
d.fill_time,
);
&d.fill_message,
)?;
let mut data = vec![0u8; padding];
data.extend_from_slice(&d.raw);
cursor2 += d.raw.len();
@@ -1494,11 +1582,16 @@ impl FileWriter {
let actual_ds_oh_sizes2: Vec<usize> = ds_blobs2.iter().map(|b| b.oh_bytes.len()).collect();
debug_assert_eq!(actual_ds_oh_sizes, actual_ds_oh_sizes2);
// libhdf5 ends a paged file on a page boundary.
let data_end = cursor2;
if let Some(ps) = page_size {
cursor2 = cursor2.next_multiple_of(ps as usize);
}
let eof_addr2 = cursor2 as u64;
let mut buf = Vec::with_capacity(cursor2);
let sb = Superblock {
version: if page_size.is_some() { 4 } else { 3 },
version: 3,
offset_size: OFFSET_SIZE,
length_size: LENGTH_SIZE,
base_address: 0,
@@ -1510,11 +1603,18 @@ impl FileWriter {
free_space_address: None,
driver_info_address: None,
consistency_flags: 0,
superblock_extension_address: Some(u64::MAX),
superblock_extension_address: Some(if sb_ext.is_some() {
SUPERBLOCK_SIZE as u64
} else {
u64::MAX
}),
checksum: None,
page_size,
page_size: None,
};
buf.extend_from_slice(&sb.serialize());
if let Some(ref ext) = sb_ext {
buf.extend_from_slice(ext);
}
// Root group OH
let mut root_links: Vec<LinkMessage> = Vec::new();
@@ -1535,7 +1635,7 @@ impl FileWriter {
root_dl,
&root_attrs,
root_dense_blob.as_ref(),
));
)?);
if let Some(ref b) = root_link_blob {
buf.extend_from_slice(&b.blob);
}
@@ -1560,7 +1660,7 @@ impl FileWriter {
dl,
&g.attrs,
group_dense_blobs[gi].as_ref(),
));
)?);
if let Some(ref b) = link_blob {
buf.extend_from_slice(&b.blob);
}
@@ -1582,7 +1682,8 @@ impl FileWriter {
buf.extend_from_slice(&blob.data);
}
debug_assert_eq!(buf.len(), cursor2);
debug_assert_eq!(buf.len(), data_end);
buf.resize(cursor2, 0);
Ok(buf)
}
}
@@ -2156,7 +2257,8 @@ mod tests {
}
#[test]
fn file_writer_v4_superblock() {
fn file_writer_paged_file_uses_v3_superblock_and_fsinfo_extension() {
// This used to write superblock "version 4", which does not exist.
let mut fw = FileWriter::new();
fw.with_page_size(4096);
fw.create_dataset("data").with_f64_data(&[1.0, 2.0]);
@@ -2164,8 +2266,31 @@ mod tests {
let sig = signature::find_signature(&bytes).unwrap();
let sb = Superblock::parse(&bytes, sig).unwrap();
assert_eq!(sb.version, 4, "expected superblock v4");
assert_eq!(sb.page_size, Some(4096));
assert_eq!(sb.version, 3);
assert_eq!(sb.superblock_extension_address, Some(48));
assert_eq!(bytes.len() % 4096, 0);
assert_eq!(sb.eof_address, bytes.len() as u64);
let ext = ObjectHeader::parse(&bytes, 48, 8, 8).unwrap();
let fsinfo = &ext.messages[0];
assert_eq!(fsinfo.msg_type, MessageType::Unknown(0x0017));
// Byte-for-byte what HDF5 2.0 writes for fs_strategy="page",
// fs_page_size=4096.
let mut expected = vec![1u8, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0];
expected.extend_from_slice(&4096u64.to_le_bytes());
expected.extend_from_slice(&[0, 0]);
expected.extend_from_slice(&[0xff; 8]);
assert_eq!(fsinfo.data, expected);
assert_eq!(fsinfo.flags, 0x14);
assert_eq!(read_dataset_f64(&bytes, "data"), vec![1.0, 2.0]);
}
#[test]
fn file_writer_rejects_page_sizes_libhdf5_would() {
for ps in [0u32, 511, MAX_FILE_SPACE_PAGE_SIZE + 1] {
let mut fw = FileWriter::new();
fw.with_page_size(ps);
assert!(fw.finish().is_err(), "page size {ps}");
}
}
#[test]