fix(format): page Fixed Array data blocks past 1024 chunks
The Fixed Array writer always packed every element into one data block behind one checksum. Past 2^10 elements libhdf5 (and our reader) expect a paged block: a page-init bitmap after the prefix, then one checksummed page per 1024 elements. Any dataset with more than 1024 chunks and no unlimited dimension failed with "incorrect metadata checksum" in h5py, h5dump and our own reader. build_fixed_array_at now takes one Option<WrittenChunk> per array slot so later fixes can leave unallocated slots. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
osobh
@@ -499,108 +499,140 @@ fn serialize_v4_fixed_array(
|
||||
buf
|
||||
}
|
||||
|
||||
/// log2 of the elements per Fixed Array data block page (the library's
|
||||
/// default, `H5D_FARRAY_MAX_DBLK_PAGE_NELMTS_BITS`).
|
||||
const FA_PAGE_BITS: u8 = 10;
|
||||
|
||||
fn push_addr(buf: &mut Vec<u8>, addr: u64, offset_size: u8) {
|
||||
match offset_size {
|
||||
4 => buf.extend_from_slice(&(addr as u32).to_le_bytes()),
|
||||
_ => buf.extend_from_slice(&addr.to_le_bytes()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Width of the chunk-size field of a filtered chunk index element. Must
|
||||
/// match the library's `H5D_FARRAY_FILT_COMPUTE_CHUNK_SIZE_LEN` (the EA and
|
||||
/// B-tree v2 indexes use the same formula):
|
||||
/// `1 + ((log2(unfiltered chunk bytes) + 8) / 8)`, capped at 8.
|
||||
pub(crate) fn filtered_chunk_size_len(slots: &[Option<WrittenChunk>]) -> usize {
|
||||
let max_raw = slots
|
||||
.iter()
|
||||
.flatten()
|
||||
.map(|c| c.raw_size)
|
||||
.max()
|
||||
.unwrap_or(1);
|
||||
let log2_val = if max_raw <= 1 {
|
||||
0
|
||||
} else {
|
||||
63 - max_raw.leading_zeros()
|
||||
};
|
||||
(1 + ((log2_val + 8) / 8) as usize).min(8)
|
||||
}
|
||||
|
||||
/// Append one chunk index element: the chunk's address, plus its stored size
|
||||
/// and filter mask when the dataset is filtered. `None` is an unallocated
|
||||
/// chunk (undefined address, zero size and mask).
|
||||
pub(crate) fn push_index_element(
|
||||
buf: &mut Vec<u8>,
|
||||
slot: Option<&WrittenChunk>,
|
||||
offset_size: u8,
|
||||
chunk_size_bytes: Option<usize>,
|
||||
) {
|
||||
match slot {
|
||||
Some(c) => {
|
||||
push_addr(buf, c.address, offset_size);
|
||||
if let Some(n) = chunk_size_bytes {
|
||||
buf.extend_from_slice(&c.compressed_size.to_le_bytes()[..n]);
|
||||
buf.extend_from_slice(&c.filter_mask.to_le_bytes());
|
||||
}
|
||||
}
|
||||
None => {
|
||||
buf.extend(core::iter::repeat_n(0xFF, offset_size as usize));
|
||||
if let Some(n) = chunk_size_bytes {
|
||||
buf.extend(core::iter::repeat_n(0x00, n + 4));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Build a complete Fixed Array at a known absolute address.
|
||||
///
|
||||
/// `slots` holds one entry per element of the array, i.e. per chunk of the
|
||||
/// dataset's *maximum* extent in the order [`crate::chunk_grid`] defines;
|
||||
/// `None` marks a chunk that is not allocated. An array with more elements
|
||||
/// than fit in one page (`2^FA_PAGE_BITS`) gets a paged data block: a
|
||||
/// page-init bitmap after the prefix, then one checksummed page per
|
||||
/// `2^FA_PAGE_BITS` elements, the last one short (`H5FA__dblock_create`).
|
||||
pub fn build_fixed_array_at(
|
||||
chunks: &[WrittenChunk],
|
||||
slots: &[Option<WrittenChunk>],
|
||||
offset_size: u8,
|
||||
length_size: u8,
|
||||
has_filters: bool,
|
||||
fa_base_address: u64,
|
||||
) -> Vec<u8> {
|
||||
let os = offset_size as usize;
|
||||
let num_elements = chunks.len();
|
||||
|
||||
// For filtered chunks, compute chunk_size encoding width.
|
||||
// Must match the HDF5 C library's H5D_FARRAY_FILT_COMPUTE_CHUNK_SIZE_LEN macro:
|
||||
// chunk_size_len = 1 + ((H5VM_log2_gen(chunk.size) + 8) / 8)
|
||||
// where chunk.size is the unfiltered chunk size in bytes (product of all chunk dims).
|
||||
let chunk_size_bytes: usize = if has_filters {
|
||||
let max_raw = chunks.iter().map(|c| c.raw_size).max().unwrap_or(1);
|
||||
let log2_val = if max_raw <= 1 {
|
||||
0
|
||||
} else {
|
||||
63 - max_raw.leading_zeros()
|
||||
};
|
||||
let len = 1 + ((log2_val + 8) / 8) as usize;
|
||||
len.min(8)
|
||||
} else {
|
||||
0
|
||||
};
|
||||
|
||||
let elem_size = if has_filters {
|
||||
os + chunk_size_bytes + 4
|
||||
} else {
|
||||
os
|
||||
};
|
||||
let num_elements = slots.len();
|
||||
|
||||
let chunk_size_bytes = has_filters.then(|| filtered_chunk_size_len(slots));
|
||||
let elem_size = os + chunk_size_bytes.map_or(0, |n| n + 4);
|
||||
let client_id: u8 = if has_filters { 1 } else { 0 };
|
||||
|
||||
// FAHD total size
|
||||
let nelmts_field_size = length_size as usize;
|
||||
let fahd_total_size = 4 + 1 + 1 + 1 + 1 + nelmts_field_size + os + 4;
|
||||
let fahd_total_size = 4 + 1 + 1 + 1 + 1 + length_size as usize + os + 4;
|
||||
let fadb_address = fa_base_address + fahd_total_size as u64;
|
||||
|
||||
// Build FAHD
|
||||
let mut fahd = Vec::with_capacity(fahd_total_size);
|
||||
fahd.extend_from_slice(b"FAHD");
|
||||
fahd.push(0); // version
|
||||
fahd.push(client_id);
|
||||
fahd.push(elem_size as u8);
|
||||
|
||||
// max_nelmts_bits: use 10 as default (page_size = 1024), matching h5py convention
|
||||
let max_bits: u8 = 10;
|
||||
fahd.push(max_bits);
|
||||
|
||||
fahd.push(FA_PAGE_BITS);
|
||||
match length_size {
|
||||
4 => fahd.extend_from_slice(&(num_elements as u32).to_le_bytes()),
|
||||
8 => fahd.extend_from_slice(&(num_elements as u64).to_le_bytes()),
|
||||
_ => fahd.extend_from_slice(&(num_elements as u64).to_le_bytes()),
|
||||
}
|
||||
|
||||
match offset_size {
|
||||
4 => fahd.extend_from_slice(&(fadb_address as u32).to_le_bytes()),
|
||||
8 => fahd.extend_from_slice(&fadb_address.to_le_bytes()),
|
||||
_ => fahd.extend_from_slice(&fadb_address.to_le_bytes()),
|
||||
}
|
||||
|
||||
// Checksum
|
||||
push_addr(&mut fahd, fadb_address, offset_size);
|
||||
let checksum = jenkins_lookup3(&fahd);
|
||||
fahd.extend_from_slice(&checksum.to_le_bytes());
|
||||
|
||||
assert_eq!(fahd.len(), fahd_total_size);
|
||||
|
||||
// Build FADB
|
||||
// FADB prefix
|
||||
let mut fadb = Vec::new();
|
||||
fadb.extend_from_slice(b"FADB");
|
||||
fadb.push(0); // version
|
||||
fadb.push(client_id);
|
||||
push_addr(&mut fadb, fa_base_address, offset_size);
|
||||
|
||||
// header address
|
||||
match offset_size {
|
||||
4 => fadb.extend_from_slice(&(fa_base_address as u32).to_le_bytes()),
|
||||
8 => fadb.extend_from_slice(&fa_base_address.to_le_bytes()),
|
||||
_ => fadb.extend_from_slice(&fa_base_address.to_le_bytes()),
|
||||
}
|
||||
|
||||
// Element data
|
||||
for chunk in chunks {
|
||||
match offset_size {
|
||||
4 => fadb.extend_from_slice(&(chunk.address as u32).to_le_bytes()),
|
||||
8 => fadb.extend_from_slice(&chunk.address.to_le_bytes()),
|
||||
_ => fadb.extend_from_slice(&chunk.address.to_le_bytes()),
|
||||
let page_nelmts = 1usize << FA_PAGE_BITS;
|
||||
if num_elements <= page_nelmts {
|
||||
// Unpaged: the elements follow the prefix, one checksum over both.
|
||||
for slot in slots {
|
||||
push_index_element(&mut fadb, slot.as_ref(), offset_size, chunk_size_bytes);
|
||||
}
|
||||
if has_filters {
|
||||
// Write compressed size using chunk_size_bytes (variable width)
|
||||
let cs_bytes = chunk.compressed_size.to_le_bytes();
|
||||
fadb.extend_from_slice(&cs_bytes[..chunk_size_bytes]);
|
||||
fadb.extend_from_slice(&chunk.filter_mask.to_le_bytes());
|
||||
let fadb_checksum = jenkins_lookup3(&fadb);
|
||||
fadb.extend_from_slice(&fadb_checksum.to_le_bytes());
|
||||
} else {
|
||||
// Paged: every page is written, so every page-init bit is set
|
||||
// (MSB-first, as `H5VM_bit_set` packs them). The prefix and bitmap
|
||||
// share a checksum; each page carries its own.
|
||||
let npages = num_elements.div_ceil(page_nelmts);
|
||||
let mut bitmap = vec![0u8; npages.div_ceil(8)];
|
||||
for p in 0..npages {
|
||||
bitmap[p / 8] |= 0x80 >> (p % 8);
|
||||
}
|
||||
fadb.extend_from_slice(&bitmap);
|
||||
let prefix_checksum = jenkins_lookup3(&fadb);
|
||||
fadb.extend_from_slice(&prefix_checksum.to_le_bytes());
|
||||
for page in slots.chunks(page_nelmts) {
|
||||
let start = fadb.len();
|
||||
for slot in page {
|
||||
push_index_element(&mut fadb, slot.as_ref(), offset_size, chunk_size_bytes);
|
||||
}
|
||||
let page_checksum = jenkins_lookup3(&fadb[start..]);
|
||||
fadb.extend_from_slice(&page_checksum.to_le_bytes());
|
||||
}
|
||||
}
|
||||
|
||||
// FADB checksum
|
||||
let fadb_checksum = jenkins_lookup3(&fadb);
|
||||
fadb.extend_from_slice(&fadb_checksum.to_le_bytes());
|
||||
|
||||
let mut combined = fahd;
|
||||
combined.extend_from_slice(&fadb);
|
||||
combined
|
||||
@@ -734,8 +766,9 @@ pub fn build_chunked_data_from_precompressed(
|
||||
)
|
||||
} else {
|
||||
let fa_address = base_address + data_buf.len() as u64;
|
||||
let slots: Vec<Option<WrittenChunk>> = written_chunks.iter().cloned().map(Some).collect();
|
||||
let fa_bytes = build_fixed_array_at(
|
||||
&written_chunks,
|
||||
&slots,
|
||||
offset_size,
|
||||
length_size,
|
||||
pre.has_filters,
|
||||
@@ -747,7 +780,7 @@ pub fn build_chunked_data_from_precompressed(
|
||||
fa_address,
|
||||
offset_size,
|
||||
element_size as u32,
|
||||
10, // max_nelmts_bits — matches h5py convention
|
||||
FA_PAGE_BITS,
|
||||
)
|
||||
};
|
||||
|
||||
@@ -1382,7 +1415,8 @@ mod tests {
|
||||
filter_mask: 0,
|
||||
},
|
||||
];
|
||||
let fa = build_fixed_array_at(&chunks, 8, 8, false, 0x2000);
|
||||
let slots: Vec<_> = chunks.into_iter().map(Some).collect();
|
||||
let fa = build_fixed_array_at(&slots, 8, 8, false, 0x2000);
|
||||
// Should start with FAHD
|
||||
assert_eq!(&fa[0..4], b"FAHD");
|
||||
// FAHD size = 4+1+1+1+1+8+8+4 = 28
|
||||
|
||||
Reference in New Issue
Block a user