use std::collections::HashMap;
use std::io::{Read, Seek, SeekFrom};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use crate::bounded::BoundedEngine;
use crate::edit::{AppendBuilder, SpaceAccounting, WriteEngine};
use crate::element::H5Element;
use crate::type_builders::{DatasetBuilder, VL_REF_SIZE};
use crate::attribute::{extract_attributes_full, extract_attributes_full_from_source};
use crate::chunk_cache::{ChunkCache, ChunkCacheConfig, ChunkCacheStats};
use crate::compound::CompoundType;
use crate::convert::TryToUsize;
use crate::data_layout::DataLayout;
use crate::data_read;
use crate::dataspace::Dataspace;
use crate::datatype::{Datatype, ReferenceType};
use crate::error::{Error, FormatError};
use crate::file_create_properties::FileCreateProperties;
use crate::file_lock::FileLocking;
use crate::file_space_info::{FileSpaceInfo, FileSpaceStrategy};
use crate::filter_pipeline::FilterPipeline;
use crate::free_space_manager;
use crate::group_v1::GroupEntry;
use crate::group_v2;
use crate::layout_info::{Chunk, ChunkIndex, Filter, Layout};
use crate::libver::LibVer;
use crate::message_type::MessageType;
use crate::object_header::ObjectHeader;
use crate::signature;
use crate::source::{
BytesSource, MetadataCacheConfig, MetadataCachingSource, ReadSeekSource, Source,
};
use crate::superblock::Superblock;
use crate::vl_data::{self, VlenStringReadOptions};
use crate::types::{AttrValue, DType, attrs_to_map, classify_datatype};
enum Backend {
InMemory(Vec<u8>),
Streaming(Box<dyn Source + Send + Sync>),
Mirror(Box<Mutex<WriteEngine>>),
Bounded(Box<Mutex<BoundedEngine>>),
}
pub(crate) enum SourceView<'a> {
Mem(&'a [u8]),
Stream(&'a (dyn Source + Send + Sync)),
}
impl Source for SourceView<'_> {
fn len(&self) -> u64 {
match self {
SourceView::Mem(b) => b.len() as u64,
SourceView::Stream(s) => s.len(),
}
}
fn read_at(&self, offset: u64, buf: &mut [u8]) -> Result<(), FormatError> {
match self {
SourceView::Mem(b) => BytesSource::new(*b).read_at(offset, buf),
SourceView::Stream(s) => s.read_at(offset, buf),
}
}
fn read_metadata_at(&self, offset: u64, len: usize) -> Result<Vec<u8>, FormatError> {
match self {
SourceView::Mem(b) => BytesSource::new(*b).read_metadata_at(offset, len),
SourceView::Stream(s) => s.read_metadata_at(offset, len),
}
}
}
struct BaseOffsetSource<'a, S: Source + ?Sized> {
inner: &'a S,
base: u64,
}
impl<S: Source + ?Sized> Source for BaseOffsetSource<'_, S> {
fn len(&self) -> u64 {
self.inner.len().saturating_sub(self.base)
}
fn read_at(&self, offset: u64, buf: &mut [u8]) -> Result<(), FormatError> {
let abs = offset
.checked_add(self.base)
.ok_or(FormatError::OffsetOverflow {
offset,
length: buf.len() as u64,
})?;
self.inner.read_at(abs, buf)
}
fn read_metadata_at(&self, offset: u64, len: usize) -> Result<Vec<u8>, FormatError> {
let abs = offset
.checked_add(self.base)
.ok_or(FormatError::OffsetOverflow {
offset,
length: len as u64,
})?;
self.inner.read_metadata_at(abs, len)
}
}
fn frame(bytes: &[u8], base: u64) -> Result<&[u8], FormatError> {
if base == 0 {
return Ok(bytes);
}
let start = base.to_usize()?;
bytes.get(start..).ok_or(FormatError::UnexpectedEof {
expected: start,
available: bytes.len(),
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[doc(alias = "fapl")]
pub struct FileAccessProperties {
metadata_cache: MetadataCacheConfig,
chunk_cache: ChunkCacheConfig,
locking: FileLocking,
}
#[deprecated(
since = "0.26.0",
note = "renamed to `FileAccessProperties`: a type standing in for a whole HDF5 property list now carries the `Properties` suffix"
)]
pub type FileAccessOptions = FileAccessProperties;
impl FileAccessProperties {
pub const fn new() -> Self {
Self {
metadata_cache: MetadataCacheConfig::disabled(),
chunk_cache: ChunkCacheConfig::new(),
locking: FileLocking::Enabled,
}
}
#[doc(alias = "H5Pset_mdc_config")]
pub const fn with_metadata_cache(mut self, metadata_cache: MetadataCacheConfig) -> Self {
self.metadata_cache = metadata_cache;
self
}
#[doc(alias = "H5Pset_cache")]
pub const fn with_chunk_cache(mut self, chunk_cache: ChunkCacheConfig) -> Self {
self.chunk_cache = chunk_cache;
self
}
#[doc(alias = "H5Pset_file_locking")]
pub const fn with_locking(mut self, locking: FileLocking) -> Self {
self.locking = locking;
self
}
pub const fn metadata_cache(&self) -> MetadataCacheConfig {
self.metadata_cache
}
pub const fn chunk_cache(&self) -> ChunkCacheConfig {
self.chunk_cache
}
pub const fn locking(&self) -> FileLocking {
self.locking
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[doc(alias = "dapl")]
pub struct DatasetAccessProperties {
chunk_cache: Option<ChunkCacheConfig>,
}
#[deprecated(
since = "0.26.0",
note = "renamed to `DatasetAccessProperties`: a type standing in for a whole HDF5 property list now carries the `Properties` suffix"
)]
pub type DatasetAccessOptions = DatasetAccessProperties;
impl DatasetAccessProperties {
pub const fn new() -> Self {
Self { chunk_cache: None }
}
#[doc(alias = "H5Pset_chunk_cache")]
pub const fn with_chunk_cache(mut self, chunk_cache: ChunkCacheConfig) -> Self {
self.chunk_cache = Some(chunk_cache);
self
}
pub const fn chunk_cache(&self) -> Option<ChunkCacheConfig> {
self.chunk_cache
}
const fn resolved_chunk_cache(&self, default: ChunkCacheConfig) -> ChunkCacheConfig {
match self.chunk_cache {
Some(config) => config,
None => default,
}
}
}
pub fn is_hdf5<P: AsRef<std::path::Path>>(path: P) -> std::io::Result<bool> {
let handle = std::fs::File::open(path)?;
let source = ReadSeekSource::new(handle).map_err(std::io::Error::other)?;
match signature::find_signature_in(&source) {
Ok(_) => Ok(true),
Err(FormatError::SignatureNotFound) => Ok(false),
Err(e) => Err(std::io::Error::other(e)),
}
}
pub fn is_hdf5_bytes(data: &[u8]) -> bool {
signature::find_signature(data).is_ok()
}
struct FileInner {
backend: Backend,
superblock: Superblock,
addr_offset: u64,
handle: Option<std::fs::File>,
file_space_info: Option<FileSpaceInfo>,
access_properties: FileAccessProperties,
closed: AtomicBool,
swmr_write: bool,
}
impl Drop for FileInner {
fn drop(&mut self) {
if self.closed.load(Ordering::Acquire) {
return;
}
match &self.backend {
Backend::Mirror(m) if self.swmr_write => {
let mut session = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let _ = session.set_consistency_flags(0);
}
Backend::Bounded(m) => {
let mut engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let _ = engine.finalize_persist();
let _ = engine.sync();
}
_ => {}
}
}
}
impl FileInner {
pub fn open<P: AsRef<std::path::Path>>(path: P) -> Result<Self, Error> {
Self::open_with_options(path, FileAccessProperties::new())
}
pub fn open_with_options<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
let bytes = std::fs::read(path.as_ref()).map_err(Error::Io)?;
Self::from_bytes_with_options(bytes, properties)
}
pub fn open_streaming<P: AsRef<std::path::Path>>(path: P) -> Result<Self, Error> {
Self::open_streaming_with_options(path, FileAccessProperties::new())
}
pub fn open_streaming_with_options<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
let handle = std::fs::File::open(path.as_ref()).map_err(Error::Io)?;
let source = ReadSeekSource::new(handle).map_err(Error::Format)?;
let source: Box<dyn Source + Send + Sync> = if properties.metadata_cache.is_enabled() {
Box::new(MetadataCachingSource::new(
source,
properties.metadata_cache,
))
} else {
Box::new(source)
};
let (superblock, addr_offset) = Self::parse_superblock_source(source.as_ref())?;
Ok(Self::from_parts(
Backend::Streaming(source),
superblock,
addr_offset,
None,
properties,
))
}
pub fn open_swmr<P: AsRef<std::path::Path>>(path: P) -> Result<Self, Error> {
Self::open_swmr_with_options(path, FileAccessProperties::new())
}
pub fn open_swmr_with_options<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
let mut handle = std::fs::File::open(path.as_ref()).map_err(Error::Io)?;
let mut data = Vec::new();
handle.read_to_end(&mut data).map_err(Error::Io)?;
let (superblock, addr_offset) = Self::parse_superblock(&data)?;
Ok(Self::from_parts(
Backend::InMemory(data),
superblock,
addr_offset,
Some(handle),
properties,
))
}
pub fn from_bytes(data: Vec<u8>) -> Result<Self, Error> {
Self::from_bytes_with_options(data, FileAccessProperties::new())
}
pub fn from_bytes_with_options(
data: Vec<u8>,
properties: FileAccessProperties,
) -> Result<Self, Error> {
let (superblock, addr_offset) = Self::parse_superblock(&data)?;
Ok(Self::from_parts(
Backend::InMemory(data),
superblock,
addr_offset,
None,
properties,
))
}
fn open_rw<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
Self::from_rw_session(
WriteEngine::open_with_locking(path, properties.locking)?,
properties,
)
}
fn from_rw_session(
session: WriteEngine,
properties: FileAccessProperties,
) -> Result<Self, Error> {
let (superblock, addr_offset) = Self::parse_superblock(session.mirror_bytes())?;
Ok(Self::from_parts(
Backend::Mirror(Box::new(Mutex::new(session))),
superblock,
addr_offset,
None,
properties,
))
}
fn open_swmr_writer<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
let mut inner = Self::from_rw_session(WriteEngine::open_swmr_writer(path)?, properties)?;
inner.swmr_write = true;
Ok(inner)
}
fn open_rw_bounded<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
let engine =
BoundedEngine::open(path.as_ref(), properties.metadata_cache, properties.locking)?;
let superblock = engine.store().superblock().clone();
Ok(Self::from_parts(
Backend::Bounded(Box::new(Mutex::new(engine))),
superblock,
0,
None,
properties,
))
}
fn check_mutable(&self, staged: bool) -> Result<(), Error> {
if self.closed.load(Ordering::Acquire) {
return Err(Error::FileClosed);
}
if staged && self.swmr_write {
return Err(Error::SwmrStagedUnsupported);
}
Ok(())
}
fn check_staged_writable(&self) -> Result<(), Error> {
match &self.backend {
Backend::Mirror(_) => self.check_mutable(true),
Backend::Bounded(_) => Err(Error::BoundedStagedUnsupported),
_ => Err(Error::ReadOnly),
}
}
pub(crate) fn source(&self) -> SourceView<'_> {
match &self.backend {
Backend::InMemory(v) => SourceView::Mem(v),
Backend::Streaming(s) => SourceView::Stream(s.as_ref()),
Backend::Mirror(_) | Backend::Bounded(_) => SourceView::Mem(&[]),
}
}
pub(crate) fn with_source<R>(&self, f: impl FnOnce(&dyn Source) -> R) -> R {
match &self.backend {
Backend::InMemory(v) => f(&BytesSource::new(v.as_slice())),
Backend::Streaming(s) => f(s.as_ref()),
Backend::Mirror(m) => {
let core = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
f(&BytesSource::new(core.mirror_bytes()))
}
Backend::Bounded(m) => {
let engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
f(engine.store())
}
}
}
fn parse_superblock(data: &[u8]) -> Result<(Superblock, u64), Error> {
let sig_offset = signature::find_signature(data)?;
let mut superblock = Superblock::parse(data, sig_offset)?;
let addr_offset = superblock.base_address;
superblock.root_group_address = superblock
.root_group_address
.checked_add(addr_offset)
.ok_or(FormatError::OffsetOverflow {
offset: superblock.root_group_address,
length: addr_offset,
})?;
debug_assert!(superblock.root_group_address >= addr_offset);
Ok((superblock, addr_offset))
}
fn parse_superblock_source<S: Source + ?Sized>(source: &S) -> Result<(Superblock, u64), Error> {
let sig_offset = signature::find_signature_in(source)?;
let mut superblock = Superblock::parse_from_source(source, sig_offset)?;
let addr_offset = superblock.base_address;
superblock.root_group_address = superblock
.root_group_address
.checked_add(addr_offset)
.ok_or(FormatError::OffsetOverflow {
offset: superblock.root_group_address,
length: addr_offset,
})?;
debug_assert!(superblock.root_group_address >= addr_offset);
Ok((superblock, addr_offset))
}
fn from_parts(
backend: Backend,
superblock: Superblock,
addr_offset: u64,
handle: Option<std::fs::File>,
access_properties: FileAccessProperties,
) -> Self {
let mut file = FileInner {
backend,
superblock,
addr_offset,
handle,
file_space_info: None,
access_properties,
closed: AtomicBool::new(false),
swmr_write: false,
};
file.file_space_info = file.read_file_space_info();
file
}
fn read_file_space_info(&self) -> Option<FileSpaceInfo> {
let rel = self.superblock.superblock_extension_address?;
if rel == u64::MAX {
return None;
}
let abs = self.addr_offset.checked_add(rel)?;
let header = self.parse_header(abs).ok()?;
let msg = header
.messages
.iter()
.find(|m| m.msg_type == MessageType::FileSpaceInfo)?;
FileSpaceInfo::parse(
&msg.data,
self.superblock.offset_size,
self.superblock.length_size,
)
.ok()
}
pub fn refresh(&mut self) -> Result<(), Error> {
let handle = self.handle.as_mut().ok_or(Error::SwmrUnsupported)?;
const MAX_ATTEMPTS: u32 = 100;
let mut last_err = None;
for attempt in 0..MAX_ATTEMPTS {
let mut data = Vec::new();
handle.seek(SeekFrom::Start(0)).map_err(Error::Io)?;
handle.read_to_end(&mut data).map_err(Error::Io)?;
match Self::parse_superblock(&data) {
Ok((superblock, addr_offset)) => {
self.backend = Backend::InMemory(data);
self.superblock = superblock;
self.addr_offset = addr_offset;
self.file_space_info = self.read_file_space_info();
return Ok(());
}
Err(e) => {
last_err = Some(e);
if attempt + 1 < MAX_ATTEMPTS {
std::thread::sleep(std::time::Duration::from_micros(
50 * (attempt + 1) as u64,
));
}
}
}
}
Err(last_err.expect("refresh retried at least once before failing"))
}
fn resolve_path(&self, path: &str) -> Result<u64, Error> {
Ok(match &self.backend {
Backend::InMemory(v) => group_v2::resolve_path_any(v, &self.superblock, path)?,
Backend::Streaming(s) => {
group_v2::resolve_path_any_from_source(s.as_ref(), &self.superblock, path)?
}
Backend::Mirror(m) => {
let core = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let data = core.mirror_bytes();
let (sb, _base) = Self::parse_superblock(data)?;
group_v2::resolve_path_any(data, &sb, path)?
}
Backend::Bounded(m) => {
let engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
group_v2::resolve_path_any_from_source(engine.store(), &self.superblock, path)?
}
})
}
fn mirror_root_address(&self) -> u64 {
if let Backend::Mirror(m) = &self.backend {
let core = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
if let Ok((sb, _base)) = Self::parse_superblock(core.mirror_bytes()) {
return sb.root_group_address;
}
}
self.superblock.root_group_address
}
pub fn as_bytes(&self) -> &[u8] {
match &self.backend {
Backend::InMemory(v) => v,
Backend::Streaming(_) | Backend::Mirror(_) | Backend::Bounded(_) => &[],
}
}
pub const fn access_properties(&self) -> FileAccessProperties {
self.access_properties
}
pub fn superblock(&self) -> &Superblock {
&self.superblock
}
pub(crate) fn in_memory_image(&self) -> Option<&[u8]> {
match &self.backend {
Backend::InMemory(data) => Some(data),
Backend::Streaming(_) | Backend::Mirror(_) | Backend::Bounded(_) => None,
}
}
pub(crate) fn base_address(&self) -> u64 {
self.addr_offset
}
pub fn file_space_strategy(&self) -> Option<FileSpaceStrategy> {
self.file_space_info.as_ref().map(|info| info.strategy)
}
pub fn file_space_info(&self) -> Option<&FileSpaceInfo> {
self.file_space_info.as_ref()
}
pub fn persisted_free_space(&self) -> Vec<(u64, u64)> {
let Some(info) = &self.file_space_info else {
return Vec::new();
};
if !info.persist {
return Vec::new();
}
let Backend::InMemory(data) = &self.backend else {
return Vec::new();
};
let mut sections = free_space_manager::read_persisted_sections(
data,
&info.manager_addrs,
self.addr_offset,
self.superblock.offset_size,
)
.unwrap_or_default();
sections.sort_by_key(|s| s.addr);
sections.into_iter().map(|s| (s.addr, s.size)).collect()
}
pub fn file_size(&self) -> u64 {
match &self.backend {
Backend::Mirror(m) => {
let core = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
core.mirror_bytes().len() as u64
}
Backend::Bounded(m) => {
let engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
engine.store().len()
}
_ => self.source().len(),
}
}
pub fn libver_bound(&self) -> LibVer {
LibVer::from_superblock_version(self.superblock.version)
}
fn parse_header(&self, address: u64) -> Result<ObjectHeader, FormatError> {
let os = self.superblock.offset_size;
let ls = self.superblock.length_size;
match &self.backend {
Backend::InMemory(v) => {
ObjectHeader::parse_with_base(v, address.to_usize()?, os, ls, self.addr_offset)
}
Backend::Streaming(s) => {
ObjectHeader::parse_from_source(s.as_ref(), address, os, ls, self.addr_offset)
}
Backend::Mirror(m) => {
let core = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
ObjectHeader::parse_with_base(
core.mirror_bytes(),
address.to_usize()?,
os,
ls,
self.addr_offset,
)
}
Backend::Bounded(m) => {
let engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
ObjectHeader::parse_from_source(engine.store(), address, os, ls, self.addr_offset)
}
}
}
fn object_at_relative(file: &Arc<FileInner>, rel_addr: u64) -> Result<Object, Error> {
if rel_addr == u64::MAX || rel_addr == 0 {
return Err(FormatError::InvalidObjectReference(rel_addr).into());
}
let abs = rel_addr
.checked_add(file.addr_offset)
.ok_or(FormatError::InvalidObjectReference(rel_addr))?;
let hdr = file.parse_header(abs)?;
if has_message(&hdr, MessageType::DataLayout) {
let chunk_cache = DatasetAccessProperties::new()
.resolved_chunk_cache(file.access_properties.chunk_cache);
Ok(Object::Dataset(Box::new(Dataset {
file: file.clone(),
address: abs,
header: hdr,
chunk_cache: ChunkCache::with_config(chunk_cache),
chunk_cache_config: chunk_cache,
path: None,
})))
} else if is_group(&hdr) {
Ok(Object::Group(Group {
file: file.clone(),
address: abs,
path: None,
}))
} else {
Err(FormatError::InvalidObjectReference(rel_addr).into())
}
}
fn offset_size(&self) -> u8 {
self.superblock.offset_size
}
fn length_size(&self) -> u8 {
self.superblock.length_size
}
fn group_children(&self, hdr: &ObjectHeader) -> Result<Vec<GroupEntry>, Error> {
let (os, ls, base) = (self.offset_size(), self.length_size(), self.addr_offset);
let mut entries = match &self.backend {
Backend::InMemory(v) => group_v2::resolve_group_entries(v, hdr, os, ls, base),
Backend::Streaming(s) => {
group_v2::resolve_group_entries_from_source(s.as_ref(), hdr, os, ls, base)
}
Backend::Mirror(m) => {
let core = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
group_v2::resolve_group_entries(core.mirror_bytes(), hdr, os, ls, base)
}
Backend::Bounded(m) => {
let engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
group_v2::resolve_group_entries_from_source(engine.store(), hdr, os, ls, base)
}
}
.map_err(Error::Format)?;
for entry in &mut entries {
entry.object_header_address = entry.object_header_address.checked_add(base).ok_or(
FormatError::OffsetOverflow {
offset: entry.object_header_address,
length: base,
},
)?;
}
Ok(entries)
}
fn attrs_of(&self, hdr: &ObjectHeader) -> Result<HashMap<String, AttrValue>, Error> {
let (os, ls, base) = (self.offset_size(), self.length_size(), self.addr_offset);
let attr_msgs = self.attr_messages_of(hdr)?;
match &self.backend {
Backend::Mirror(m) => {
let core = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
Ok(attrs_to_map(
&attr_msgs,
&BytesSource::new(core.mirror_bytes()),
os,
ls,
base,
))
}
Backend::Bounded(m) => {
let engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
Ok(attrs_to_map(&attr_msgs, engine.store(), os, ls, base))
}
_ => Ok(attrs_to_map(&attr_msgs, &self.source(), os, ls, base)),
}
}
pub(crate) fn attr_message_names_of(&self, hdr: &ObjectHeader) -> Result<Vec<String>, Error> {
Ok(self
.attr_messages_of(hdr)?
.into_iter()
.map(|a| a.name)
.collect())
}
fn attr_messages_of(
&self,
hdr: &ObjectHeader,
) -> Result<Vec<crate::attribute::AttributeMessage>, Error> {
let (os, ls) = (self.offset_size(), self.length_size());
let base = self.addr_offset;
match &self.backend {
Backend::InMemory(v) => Ok(extract_attributes_full(frame(v, base)?, hdr, os, ls)?),
Backend::Streaming(s) if base == 0 => Ok(extract_attributes_full_from_source(
s.as_ref(),
hdr,
os,
ls,
)?),
Backend::Streaming(s) => {
let framed = BaseOffsetSource {
inner: s.as_ref(),
base,
};
Ok(extract_attributes_full_from_source(&framed, hdr, os, ls)?)
}
Backend::Mirror(m) => {
let core = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
Ok(extract_attributes_full(
frame(core.mirror_bytes(), base)?,
hdr,
os,
ls,
)?)
}
Backend::Bounded(m) => {
let engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let store = engine.store();
if base == 0 {
Ok(extract_attributes_full_from_source(store, hdr, os, ls)?)
} else {
let framed = BaseOffsetSource { inner: store, base };
Ok(extract_attributes_full_from_source(&framed, hdr, os, ls)?)
}
}
}
}
fn read_dataset_raw(
&self,
dl: &DataLayout,
ds: &Dataspace,
dt: &Datatype,
pipeline: Option<&FilterPipeline>,
cache: &ChunkCache,
) -> Result<Vec<u8>, FormatError> {
let (os, ls) = (self.offset_size(), self.length_size());
let base = self.addr_offset;
match &self.backend {
Backend::InMemory(v) => data_read::read_raw_data_cached(
frame(v, base)?,
dl,
ds,
dt,
pipeline,
os,
ls,
cache,
),
Backend::Streaming(s) if base == 0 => data_read::read_raw_data_cached_from_source(
s.as_ref(),
dl,
ds,
dt,
pipeline,
os,
ls,
cache,
),
Backend::Streaming(s) => {
let framed = BaseOffsetSource {
inner: s.as_ref(),
base,
};
data_read::read_raw_data_cached_from_source(
&framed, dl, ds, dt, pipeline, os, ls, cache,
)
}
Backend::Mirror(m) => {
let core = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let data = core.mirror_bytes();
let frame = if base == 0 {
data
} else {
let start = base.to_usize()?;
data.get(start..).ok_or(FormatError::UnexpectedEof {
expected: start,
available: data.len(),
})?
};
data_read::read_raw_data_cached(frame, dl, ds, dt, pipeline, os, ls, cache)
}
Backend::Bounded(m) => {
debug_assert_eq!(base, 0);
let engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
data_read::read_raw_data_cached_from_source(
engine.store(),
dl,
ds,
dt,
pipeline,
os,
ls,
cache,
)
}
}
}
#[allow(clippy::too_many_arguments)]
fn read_dataset_raw_rows(
&self,
dl: &DataLayout,
ds: &Dataspace,
dt: &Datatype,
pipeline: Option<&FilterPipeline>,
cache: &ChunkCache,
start_row: u64,
num_rows: u64,
) -> Result<Vec<u8>, FormatError> {
let (os, ls) = (self.offset_size(), self.length_size());
let elem_size = dt.type_size() as usize;
let row_elems: usize = ds.dimensions.iter().skip(1).try_fold(1usize, |acc, &d| {
acc.checked_mul(d.to_usize()?)
.ok_or(FormatError::OffsetOverflow {
offset: acc as u64,
length: d,
})
})?;
let row_bytes = row_elems
.checked_mul(elem_size)
.ok_or(FormatError::OffsetOverflow {
offset: row_elems as u64,
length: elem_size as u64,
})?;
if let DataLayout::Compact { data } = dl {
let start = start_row.to_usize()?.checked_mul(row_bytes);
let len = num_rows.to_usize()?.checked_mul(row_bytes);
let (Some(start), Some(len)) = (start, len) else {
return Err(FormatError::OffsetOverflow {
offset: start_row,
length: row_bytes as u64,
});
};
let end = start.checked_add(len).ok_or(FormatError::OffsetOverflow {
offset: start as u64,
length: len as u64,
})?;
return data
.get(start..end)
.map(<[u8]>::to_vec)
.ok_or(FormatError::DataSizeMismatch {
expected: end,
actual: data.len(),
});
}
let base = self.addr_offset;
match &self.backend {
Backend::InMemory(v) => {
let frame = if base == 0 {
v.as_slice()
} else {
let start = base.to_usize()?;
v.get(start..).ok_or(FormatError::UnexpectedEof {
expected: start,
available: v.len(),
})?
};
read_rows_framed(
&BytesSource::new(frame),
dl,
ds,
dt,
pipeline,
os,
ls,
cache,
start_row,
num_rows,
row_bytes,
)
}
Backend::Streaming(s) if base == 0 => read_rows_framed(
s.as_ref(),
dl,
ds,
dt,
pipeline,
os,
ls,
cache,
start_row,
num_rows,
row_bytes,
),
Backend::Streaming(s) => {
let framed = BaseOffsetSource {
inner: s.as_ref(),
base,
};
read_rows_framed(
&framed, dl, ds, dt, pipeline, os, ls, cache, start_row, num_rows, row_bytes,
)
}
Backend::Mirror(m) => {
let core = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let data = core.mirror_bytes();
let frame = if base == 0 {
data
} else {
let start = base.to_usize()?;
data.get(start..).ok_or(FormatError::UnexpectedEof {
expected: start,
available: data.len(),
})?
};
read_rows_framed(
&BytesSource::new(frame),
dl,
ds,
dt,
pipeline,
os,
ls,
cache,
start_row,
num_rows,
row_bytes,
)
}
Backend::Bounded(m) => {
debug_assert_eq!(base, 0);
let engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
read_rows_framed(
engine.store(),
dl,
ds,
dt,
pipeline,
os,
ls,
cache,
start_row,
num_rows,
row_bytes,
)
}
}
}
}
#[allow(clippy::too_many_arguments)]
fn read_rows_framed<S: Source + ?Sized>(
source: &S,
dl: &DataLayout,
ds: &Dataspace,
dt: &Datatype,
pipeline: Option<&FilterPipeline>,
os: u8,
ls: u8,
cache: &ChunkCache,
start_row: u64,
num_rows: u64,
row_bytes: usize,
) -> Result<Vec<u8>, FormatError> {
if num_rows == 0 && !matches!(dl, DataLayout::Virtual { .. }) {
return Ok(Vec::new());
}
match dl {
DataLayout::Compact { .. } => unreachable!("compact is handled before framing"),
DataLayout::Contiguous { address, size } => {
let addr = address.ok_or(FormatError::NoDataAllocated)?;
let start =
start_row
.checked_mul(row_bytes as u64)
.ok_or(FormatError::OffsetOverflow {
offset: start_row,
length: row_bytes as u64,
})?;
let len =
num_rows
.to_usize()?
.checked_mul(row_bytes)
.ok_or(FormatError::OffsetOverflow {
offset: num_rows,
length: row_bytes as u64,
})?;
if start.saturating_add(len as u64) > *size {
return Err(FormatError::DataSizeMismatch {
expected: start.to_usize()?.saturating_add(len),
actual: (*size).to_usize()?,
});
}
let off = addr.checked_add(start).ok_or(FormatError::OffsetOverflow {
offset: addr,
length: start,
})?;
source.read_exact_at(off, len)
}
DataLayout::Chunked { .. } => {
match crate::chunked_read::read_chunked_rows_from_source(
source, dl, ds, dt, pipeline, os, ls, cache, start_row, num_rows,
)? {
Some(bytes) => Ok(bytes),
None => {
let full = data_read::read_raw_data_cached_from_source(
source, dl, ds, dt, pipeline, os, ls, cache,
)?;
let start = start_row.to_usize()? * row_bytes;
let len = num_rows.to_usize()? * row_bytes;
full.get(start..start + len).map(<[u8]>::to_vec).ok_or(
FormatError::DataSizeMismatch {
expected: start + len,
actual: full.len(),
},
)
}
}
}
DataLayout::Virtual { .. } => Err(FormatError::UnsupportedVirtualLayout),
}
}
impl std::fmt::Debug for FileInner {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("File")
.field("size", &self.file_size())
.field("superblock_version", &self.superblock.version)
.finish()
}
}
#[derive(Clone)]
pub struct File {
inner: Arc<FileInner>,
}
impl std::fmt::Debug for File {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
std::fmt::Debug::fmt(&*self.inner, f)
}
}
impl File {
pub fn open<P: AsRef<std::path::Path>>(path: P) -> Result<Self, Error> {
Ok(File {
inner: Arc::new(FileInner::open(path)?),
})
}
pub fn open_with_options<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
Ok(File {
inner: Arc::new(FileInner::open_with_options(path, properties)?),
})
}
pub fn open_streaming<P: AsRef<std::path::Path>>(path: P) -> Result<Self, Error> {
Ok(File {
inner: Arc::new(FileInner::open_streaming(path)?),
})
}
pub fn open_streaming_with_options<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
Ok(File {
inner: Arc::new(FileInner::open_streaming_with_options(path, properties)?),
})
}
pub fn open_swmr<P: AsRef<std::path::Path>>(path: P) -> Result<Self, Error> {
Ok(File {
inner: Arc::new(FileInner::open_swmr(path)?),
})
}
pub fn open_swmr_with_options<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
Ok(File {
inner: Arc::new(FileInner::open_swmr_with_options(path, properties)?),
})
}
pub fn from_bytes(data: Vec<u8>) -> Result<Self, Error> {
Ok(File {
inner: Arc::new(FileInner::from_bytes(data)?),
})
}
pub fn from_bytes_with_options(
data: Vec<u8>,
properties: FileAccessProperties,
) -> Result<Self, Error> {
Ok(File {
inner: Arc::new(FileInner::from_bytes_with_options(data, properties)?),
})
}
#[doc(alias = "H5Fopen")]
pub fn open_rw<P: AsRef<std::path::Path>>(path: P) -> Result<Self, Error> {
Self::open_rw_with_options(path, FileAccessProperties::new())
}
pub fn open_rw_with_options<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
Ok(File {
inner: Arc::new(FileInner::open_rw(path, properties)?),
})
}
#[doc(alias = "H5F_ACC_SWMR_WRITE")]
pub fn open_swmr_writer<P: AsRef<std::path::Path>>(path: P) -> Result<Self, Error> {
Self::open_swmr_writer_with_options(path, FileAccessProperties::new())
}
pub fn open_swmr_writer_with_options<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
Ok(File {
inner: Arc::new(FileInner::open_swmr_writer(path, properties)?),
})
}
pub fn open_rw_bounded<P: AsRef<std::path::Path>>(path: P) -> Result<Self, Error> {
Self::open_rw_bounded_with_options(path, FileAccessProperties::new())
}
pub fn open_rw_bounded_with_options<P: AsRef<std::path::Path>>(
path: P,
properties: FileAccessProperties,
) -> Result<Self, Error> {
Ok(File {
inner: Arc::new(FileInner::open_rw_bounded(path, properties)?),
})
}
pub fn clear_swmr_flag<P: AsRef<std::path::Path>>(path: P) -> Result<(), Error> {
crate::file_lock::clear_swmr_flag_at(path.as_ref())
}
#[doc(alias = "H5Fcreate")]
pub fn create<P: AsRef<std::path::Path>>(path: P) -> Result<Self, Error> {
Self::create_with_options(
path,
FileCreateProperties::new(),
FileAccessProperties::new(),
)
}
pub fn create_with_options<P: AsRef<std::path::Path>>(
path: P,
create: FileCreateProperties,
access: FileAccessProperties,
) -> Result<Self, Error> {
let mut builder = crate::writer::FileBuilder::new();
builder.with_create_properties(create);
let bytes = builder.finish()?;
std::fs::write(path.as_ref(), bytes).map_err(Error::Io)?;
Self::open_rw_with_options(path, access)
}
pub fn commit(&self) -> Result<(), Error> {
self.with_mirror_session(true, |session| session.commit())
}
pub fn copy(&self, src: &str, dst: &str) -> Result<(), Error> {
self.with_mirror_session(true, |session| {
session.copy(&normalize_path(src), &normalize_path(dst));
Ok(())
})
}
pub fn copy_from(&self, source: &File, src: &str, dst: &str) -> Result<(), Error> {
self.with_mirror_session(true, |session| session.copy_from(source, src, dst))
}
pub fn has_staged_edits(&self) -> bool {
match &self.inner.backend {
Backend::Mirror(m) => {
let session = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
session.has_staged_edits()
}
_ => false,
}
}
pub fn space_accounting(&self) -> Result<SpaceAccounting, Error> {
match &self.inner.backend {
Backend::Mirror(m) => {
let session = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
Ok(session.space_accounting())
}
Backend::Bounded(_) => Err(Error::BoundedStagedUnsupported),
_ => Err(Error::ReadOnly),
}
}
pub fn close(self) -> Result<(), Error> {
if let Backend::Bounded(m) = &self.inner.backend {
let mut engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
engine.finalize_persist()?;
engine.sync()?;
drop(engine);
self.inner.closed.store(true, Ordering::Release);
return Ok(());
}
if matches!(self.inner.backend, Backend::Mirror(_)) {
if self.inner.swmr_write {
if let Backend::Mirror(m) = &self.inner.backend {
let mut session = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
session.set_consistency_flags(0)?;
}
} else {
self.commit()?;
}
self.inner.closed.store(true, Ordering::Release);
}
Ok(())
}
fn with_mirror_session<R>(
&self,
staged: bool,
f: impl FnOnce(&mut WriteEngine) -> Result<R, Error>,
) -> Result<R, Error> {
let Backend::Mirror(m) = &self.inner.backend else {
if matches!(self.inner.backend, Backend::Bounded(_)) {
return Err(Error::BoundedStagedUnsupported);
}
return Err(Error::ReadOnly);
};
self.inner.check_mutable(staged)?;
let mut session = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
f(&mut session)
}
pub fn root(&self) -> Group {
Group {
address: self.inner.mirror_root_address(),
file: self.inner.clone(),
path: Some(String::new()),
}
}
pub fn dataset(&self, path: &str) -> Result<Dataset, Error> {
self.dataset_with_options(path, DatasetAccessProperties::new())
}
pub fn dataset_with_options(
&self,
path: &str,
properties: DatasetAccessProperties,
) -> Result<Dataset, Error> {
let addr = self.inner.resolve_path(path)?;
let hdr = self.inner.parse_header(addr)?;
if !has_message(&hdr, MessageType::DataLayout) {
return Err(Error::NotADataset(path.to_string()));
}
let chunk_cache = properties.resolved_chunk_cache(self.inner.access_properties.chunk_cache);
Ok(Dataset {
file: self.inner.clone(),
address: addr,
header: hdr,
chunk_cache: ChunkCache::with_config(chunk_cache),
chunk_cache_config: chunk_cache,
path: Some(normalize_path(path)),
})
}
pub fn group(&self, path: &str) -> Result<Group, Error> {
let addr = self.inner.resolve_path(path)?;
Ok(Group {
file: self.inner.clone(),
address: addr,
path: Some(normalize_path(path)),
})
}
pub fn refresh(&mut self) -> Result<(), Error> {
let inner = Arc::get_mut(&mut self.inner).ok_or(Error::HandlesOutstanding)?;
inner.refresh()
}
pub fn as_bytes(&self) -> &[u8] {
self.inner.as_bytes()
}
pub fn access_properties(&self) -> FileAccessProperties {
self.inner.access_properties()
}
#[deprecated(since = "0.26.0", note = "renamed to `access_properties`")]
pub fn access_options(&self) -> FileAccessProperties {
self.access_properties()
}
pub fn superblock(&self) -> &Superblock {
self.inner.superblock()
}
pub fn file_space_strategy(&self) -> Option<FileSpaceStrategy> {
self.inner.file_space_strategy()
}
pub fn file_space_info(&self) -> Option<&FileSpaceInfo> {
self.inner.file_space_info()
}
pub fn persisted_free_space(&self) -> Vec<(u64, u64)> {
self.inner.persisted_free_space()
}
pub fn file_size(&self) -> u64 {
self.inner.file_size()
}
pub fn libver_bound(&self) -> LibVer {
self.inner.libver_bound()
}
pub(crate) fn source(&self) -> SourceView<'_> {
self.inner.source()
}
pub(crate) fn in_memory_image(&self) -> Option<&[u8]> {
self.inner.in_memory_image()
}
pub(crate) fn base_address(&self) -> u64 {
self.inner.base_address()
}
}
#[non_exhaustive]
pub enum Object {
Group(Group),
Dataset(Box<Dataset>),
}
impl std::fmt::Debug for Object {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Object::Group(_) => f.write_str("Object::Group"),
Object::Dataset(_) => f.write_str("Object::Dataset"),
}
}
}
pub struct StagedGroup<'a> {
ops: &'a mut Vec<StagedOp>,
path: String,
}
impl StagedGroup<'_> {
pub fn set_attr(&mut self, name: &str, value: AttrValue) -> &mut Self {
self.ops.push(StagedOp::SetGroupAttr {
path: self.path.clone(),
name: name.to_string(),
value,
});
self
}
pub fn create_group(&mut self, name: &str) -> &mut Self {
self.create_group_with(name, |_| {})
}
pub fn create_group_with(
&mut self,
name: &str,
build: impl FnOnce(&mut StagedGroup<'_>),
) -> &mut Self {
let child = format!("{}/{}", self.path, name);
self.ops.push(StagedOp::CreateGroup(child.clone()));
let mut staged = StagedGroup {
ops: &mut *self.ops,
path: child,
};
build(&mut staged);
self
}
pub fn create_dataset(
&mut self,
name: &str,
build: impl FnOnce(&mut DatasetBuilder),
) -> &mut Self {
let mut builder = DatasetBuilder::new(name);
build(&mut builder);
self.ops.push(StagedOp::CreateDataset {
path: format!("{}/{}", self.path, name),
builder: Box::new(builder),
});
self
}
}
enum StagedOp {
CreateGroup(String),
SetGroupAttr {
path: String,
name: String,
value: AttrValue,
},
CreateDataset {
path: String,
builder: Box<DatasetBuilder>,
},
}
impl StagedOp {
fn apply(self, session: &mut WriteEngine) {
match self {
StagedOp::CreateGroup(path) => session.create_group(&path),
StagedOp::SetGroupAttr { path, name, value } => {
session.set_group_attr(&path, &name, value);
}
StagedOp::CreateDataset { path, builder } => {
session.stage_created_dataset(&path, *builder);
}
}
}
}
pub struct Group {
file: Arc<FileInner>,
address: u64,
path: Option<String>,
}
impl Group {
pub(crate) fn header_address(&self) -> u64 {
self.address
}
pub fn datasets(&self) -> Result<Vec<String>, Error> {
let entries = self.children()?;
let mut names = Vec::new();
for entry in &entries {
let hdr = self.file.parse_header(entry.object_header_address)?;
if has_message(&hdr, MessageType::DataLayout) {
names.push(entry.name.clone());
}
}
Ok(names)
}
pub fn groups(&self) -> Result<Vec<String>, Error> {
let entries = self.children()?;
let mut names = Vec::new();
for entry in &entries {
let hdr = self.file.parse_header(entry.object_header_address)?;
if is_group(&hdr) {
names.push(entry.name.clone());
}
}
Ok(names)
}
pub fn attrs(&self) -> Result<HashMap<String, AttrValue>, Error> {
let hdr = self.file.parse_header(self.address)?;
self.file.attrs_of(&hdr)
}
pub(crate) fn attr_names(&self) -> Result<Vec<String>, Error> {
let hdr = self.file.parse_header(self.address)?;
self.file.attr_message_names_of(&hdr)
}
pub fn dataset(&self, name: &str) -> Result<Dataset, Error> {
self.dataset_with_options(name, DatasetAccessProperties::new())
}
pub fn dataset_with_options(
&self,
name: &str,
properties: DatasetAccessProperties,
) -> Result<Dataset, Error> {
let entries = self.children()?;
let entry = entries
.iter()
.find(|e| e.name == name)
.ok_or_else(|| Error::Format(FormatError::PathNotFound(name.to_string())))?;
let hdr = self.file.parse_header(entry.object_header_address)?;
if !has_message(&hdr, MessageType::DataLayout) {
return Err(Error::NotADataset(name.to_string()));
}
let chunk_cache = properties.resolved_chunk_cache(self.file.access_properties.chunk_cache);
Ok(Dataset {
file: self.file.clone(),
address: entry.object_header_address,
header: hdr,
chunk_cache: ChunkCache::with_config(chunk_cache),
chunk_cache_config: chunk_cache,
path: self.child_path(name),
})
}
pub fn group(&self, name: &str) -> Result<Group, Error> {
let entries = self.children()?;
let entry = entries
.iter()
.find(|e| e.name == name)
.ok_or_else(|| Error::Format(FormatError::PathNotFound(name.to_string())))?;
Ok(Group {
file: self.file.clone(),
address: entry.object_header_address,
path: self.child_path(name),
})
}
fn child_path(&self, name: &str) -> Option<String> {
self.path.as_ref().map(|p| {
if p.is_empty() {
name.to_string()
} else {
format!("{p}/{name}")
}
})
}
pub fn create_group(&self, name: &str) -> Result<(), Error> {
self.create_group_with(name, |_| {})
}
pub fn create_group_with(
&self,
name: &str,
build: impl FnOnce(&mut StagedGroup<'_>),
) -> Result<(), Error> {
let child = self.child_edit_path(name)?;
let mut ops = vec![StagedOp::CreateGroup(child.clone())];
build(&mut StagedGroup {
ops: &mut ops,
path: child,
});
self.apply_staged(ops)
}
pub fn create_dataset(
&self,
name: &str,
build: impl FnOnce(&mut DatasetBuilder),
) -> Result<(), Error> {
let child = self.child_edit_path(name)?;
let mut builder = DatasetBuilder::new(name);
build(&mut builder);
self.apply_staged(vec![StagedOp::CreateDataset {
path: child,
builder: Box::new(builder),
}])
}
pub fn delete(&self, name: &str) -> Result<(), Error> {
self.with_child_session(name, |session, child| {
session.delete(child);
Ok(())
})
}
pub fn set_attr(&self, name: &str, value: AttrValue) -> Result<(), Error> {
self.with_own_session(|session, path| {
session.set_group_attr(path, name, value);
Ok(())
})
}
pub fn remove_attr(&self, name: &str) -> Result<(), Error> {
self.with_own_session(|session, path| {
session.remove_group_attr(path, name);
Ok(())
})
}
fn with_child_session<R>(
&self,
name: &str,
f: impl FnOnce(&mut WriteEngine, &str) -> Result<R, Error>,
) -> Result<R, Error> {
let Backend::Mirror(m) = &self.file.backend else {
if matches!(self.file.backend, Backend::Bounded(_)) {
return Err(Error::BoundedStagedUnsupported);
}
return Err(Error::ReadOnly);
};
self.file.check_mutable(true)?;
let child = self.child_path(name).ok_or(Error::ReadOnly)?;
let mut session = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
f(&mut session, &child)
}
fn child_edit_path(&self, name: &str) -> Result<String, Error> {
self.file.check_staged_writable()?;
self.child_path(name).ok_or(Error::ReadOnly)
}
fn apply_staged(&self, ops: Vec<StagedOp>) -> Result<(), Error> {
self.file.check_staged_writable()?;
let Backend::Mirror(m) = &self.file.backend else {
return Err(Error::ReadOnly);
};
let mut session = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
for op in ops {
op.apply(&mut session);
}
Ok(())
}
fn with_own_session<R>(
&self,
f: impl FnOnce(&mut WriteEngine, &str) -> Result<R, Error>,
) -> Result<R, Error> {
let Backend::Mirror(m) = &self.file.backend else {
if matches!(self.file.backend, Backend::Bounded(_)) {
return Err(Error::BoundedStagedUnsupported);
}
return Err(Error::ReadOnly);
};
self.file.check_mutable(true)?;
let path = self.path.clone().ok_or(Error::ReadOnly)?;
let mut session = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
f(&mut session, &path)
}
fn children(&self) -> Result<Vec<GroupEntry>, Error> {
let hdr = self.file.parse_header(self.address)?;
self.file.group_children(&hdr)
}
}
pub struct Dataset {
file: Arc<FileInner>,
address: u64,
header: ObjectHeader,
chunk_cache: ChunkCache,
chunk_cache_config: ChunkCacheConfig,
path: Option<String>,
}
impl std::fmt::Debug for Dataset {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Dataset")
.field("messages", &self.header.messages.len())
.finish()
}
}
impl Dataset {
pub(crate) fn header_address(&self) -> u64 {
self.address
}
pub fn append<T: H5Element>(&mut self, data: &[T]) -> Result<(), Error> {
if matches!(self.file.backend, Backend::Bounded(_)) {
let g = self.bounded_geometry()?;
return self.bounded_append_batches(g, data.len() as u64, |b, r| {
b.append(&data[r]);
});
}
self.with_session_mut(false, |session, path| session.append_inplace(path, data))
}
pub fn append_raw(&mut self, bytes: &[u8]) -> Result<(), Error> {
if matches!(self.file.backend, Backend::Bounded(_)) {
let g = self.bounded_geometry()?;
let es = g.element_size.max(1);
if bytes.len() % es != 0 {
return Err(Error::AppendInPlaceUnsupported(
"appended byte length is not a whole number of elements",
));
}
let total = (bytes.len() / es) as u64;
return self.bounded_append_batches(g, total, |b, r| {
b.append_raw(&bytes[r.start * es..r.end * es]);
});
}
self.with_session_mut(false, |session, path| {
session.append_inplace_raw(path, bytes)
})
}
fn bounded_geometry(&self) -> Result<crate::bounded::AppendGeometry, Error> {
let Backend::Bounded(m) = &self.file.backend else {
return Err(Error::ReadOnly);
};
self.file.check_mutable(false)?;
let mut engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
engine.append_geometry(self.address)
}
fn bounded_append_batches(
&mut self,
g: crate::bounded::AppendGeometry,
total_elems: u64,
fill: impl Fn(&mut AppendBuilder, std::ops::Range<usize>),
) -> Result<(), Error> {
let Backend::Bounded(m) = &self.file.backend else {
return Err(Error::ReadOnly);
};
if g.filtered && (g.current_dim % g.chunk_elems != 0 || total_elems % g.chunk_elems != 0) {
return Err(Error::AppendInPlaceUnsupported(
"a filtered dataset can only be appended in place in whole chunks (the current \
length and the appended length must both be multiples of the chunk length); \
use Dataset::append_staged for a non-chunk-aligned filtered append",
));
}
let mut dim = g.current_dim;
let mut done = 0u64;
loop {
self.file.check_mutable(false)?;
let to_boundary = (g.chunk_elems - dim % g.chunk_elems) % g.chunk_elems;
let take = (total_elems - done).min(to_boundary + g.full_batch_elems);
let mut b = AppendBuilder::new();
fill(&mut b, done.to_usize()?..(done + take).to_usize()?);
{
let mut engine = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
engine.append_gathered(self.address, &b, 4)?;
}
dim += take;
done += take;
if done >= total_elems {
break;
}
}
self.header = self.file.parse_header(self.address)?;
self.chunk_cache.clear();
Ok(())
}
pub fn write<T: H5Element>(&mut self, data: &[T]) -> Result<(), Error> {
self.check_staged_edit()?;
let mut builder = DatasetBuilder::new("");
T::write_into(&mut builder, data);
self.with_session_mut(true, |session, path| {
session.stage_dataset_write(path, builder);
Ok(())
})
}
pub fn write_staged(&mut self, build: impl FnOnce(&mut DatasetBuilder)) -> Result<(), Error> {
self.check_staged_edit()?;
let mut builder = DatasetBuilder::new("");
build(&mut builder);
self.with_session_mut(true, |session, path| {
session.stage_dataset_write(path, builder);
Ok(())
})
}
pub fn append_staged(&mut self, build: impl FnOnce(&mut AppendBuilder)) -> Result<(), Error> {
self.check_staged_edit()?;
let mut builder = AppendBuilder::new();
build(&mut builder);
self.with_session_mut(true, |session, path| {
session.stage_dataset_append(path, builder);
Ok(())
})
}
pub fn set_attr(&mut self, name: &str, value: AttrValue) -> Result<(), Error> {
self.with_session_mut(true, |session, path| {
session.set_dataset_attr(path, name, value);
Ok(())
})
}
pub fn remove_attr(&mut self, name: &str) -> Result<(), Error> {
self.with_session_mut(true, |session, path| {
session.remove_dataset_attr(path, name);
Ok(())
})
}
fn check_staged_edit(&self) -> Result<(), Error> {
self.file.check_staged_writable()?;
if self.path.is_none() {
return Err(Error::ReadOnly);
}
Ok(())
}
fn with_session_mut<R>(
&mut self,
staged: bool,
f: impl FnOnce(&mut WriteEngine, &str) -> Result<R, Error>,
) -> Result<R, Error> {
let Backend::Mirror(m) = &self.file.backend else {
if matches!(self.file.backend, Backend::Bounded(_)) {
return Err(Error::BoundedStagedUnsupported);
}
return Err(Error::ReadOnly);
};
self.file.check_mutable(staged)?;
let path = self.path.clone().ok_or(Error::ReadOnly)?;
let out = {
let mut session = m.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
f(&mut session, &path)?
};
self.header = self.file.parse_header(self.address)?;
self.chunk_cache.clear();
Ok(out)
}
pub const fn chunk_cache_config(&self) -> ChunkCacheConfig {
self.chunk_cache_config
}
pub fn chunk_cache_stats(&self) -> ChunkCacheStats {
self.chunk_cache.stats()
}
pub fn shape(&self) -> Result<Vec<u64>, Error> {
let ds = self.dataspace()?;
Ok(ds.dimensions.clone())
}
pub fn maxshape(&self) -> Result<Option<Vec<u64>>, Error> {
let ds = self.dataspace()?;
match &ds.max_dimensions {
Some(md) if *md != ds.dimensions => Ok(Some(md.clone())),
_ => Ok(None),
}
}
pub fn is_chunked(&self) -> bool {
matches!(self.data_layout(), Ok(DataLayout::Chunked { .. }))
}
pub fn chunk_shape(&self) -> Result<Option<Vec<u64>>, Error> {
let DataLayout::Chunked {
chunk_dimensions, ..
} = self.data_layout()?
else {
return Ok(None);
};
let rank = self.dataspace()?.dimensions.len();
if chunk_dimensions.len() <= rank {
return Ok(None);
}
Ok(Some(
chunk_dimensions[..rank]
.iter()
.map(|&c| u64::from(c))
.collect(),
))
}
pub fn filters(&self) -> Vec<u16> {
self.filter_pipeline_parsed()
.map(|p| p.filters.iter().map(|f| f.filter_id).collect())
.unwrap_or_default()
}
pub fn layout(&self) -> Result<Layout, Error> {
Ok(match self.data_layout()? {
DataLayout::Compact { data } => Layout::Compact {
size: data.len() as u64,
},
DataLayout::Contiguous { address, size } => Layout::Contiguous {
address: self.absolute_address(address)?,
size,
},
DataLayout::Chunked {
version,
chunk_index_type,
..
} => Layout::Chunked {
chunk_shape: self.chunk_shape()?.unwrap_or_default(),
index: ChunkIndex::from_layout(version, chunk_index_type)?,
},
DataLayout::Virtual { .. } => Layout::Virtual,
})
}
pub fn chunk_index(&self) -> Result<Option<ChunkIndex>, Error> {
match self.data_layout()? {
DataLayout::Chunked {
version,
chunk_index_type,
..
} => Ok(Some(ChunkIndex::from_layout(version, chunk_index_type)?)),
_ => Ok(None),
}
}
pub fn chunks(&self) -> Result<Vec<Chunk>, Error> {
let rank = self.dataspace()?.dimensions.len();
Ok(self
.raw_chunks()?
.into_iter()
.map(|c| Chunk {
offset: c.offsets.into_iter().take(rank).collect(),
address: c.address,
storage_size: u64::from(c.chunk_size),
filter_mask: c.filter_mask,
})
.collect())
}
pub fn filter_pipeline(&self) -> Vec<Filter> {
self.filter_pipeline_parsed()
.map(|p| {
p.filters
.into_iter()
.map(|f| Filter {
id: f.filter_id,
name: f.name,
is_optional: f.flags & 0x1 != 0,
client_data: f.client_data,
})
.collect()
})
.unwrap_or_default()
}
fn absolute_address(&self, address: Option<u64>) -> Result<Option<u64>, Error> {
match address {
Some(rel) => Ok(Some(rel.checked_add(self.file.addr_offset).ok_or(
crate::error::FormatError::OffsetOverflow {
offset: rel,
length: 0,
},
)?)),
None => Ok(None),
}
}
pub fn dtype(&self) -> Result<DType, Error> {
let dt = self.datatype()?;
Ok(classify_datatype(&dt))
}
pub fn element_size(&self) -> Result<u64, Error> {
Ok(u64::from(self.datatype()?.type_size()))
}
pub(crate) fn defined_fill_bytes(&self) -> Result<Option<Vec<u8>>, Error> {
let msg = self
.header
.messages
.iter()
.find(|m| m.msg_type == MessageType::FillValue)
.or_else(|| {
self.header
.messages
.iter()
.find(|m| m.msg_type == MessageType::FillValueOld)
});
match msg {
Some(m) => Ok(crate::fill_value::parse_defined_fill_value(
m.msg_type, &m.data,
)?),
None => Ok(None),
}
}
pub fn read_f64(&self) -> Result<Vec<f64>, Error> {
let raw = self.read_raw()?;
let dt = self.datatype()?;
Ok(data_read::read_as_f64(&raw, &dt)?)
}
pub fn read_f32(&self) -> Result<Vec<f32>, Error> {
let raw = self.read_raw()?;
let dt = self.datatype()?;
Ok(data_read::read_as_f32(&raw, &dt)?)
}
pub fn read_i32(&self) -> Result<Vec<i32>, Error> {
let raw = self.read_raw()?;
let dt = self.datatype()?;
Ok(data_read::read_as_i32(&raw, &dt)?)
}
pub fn read_i64(&self) -> Result<Vec<i64>, Error> {
let raw = self.read_raw()?;
let dt = self.datatype()?;
Ok(data_read::read_as_i64(&raw, &dt)?)
}
pub fn read_u64(&self) -> Result<Vec<u64>, Error> {
let raw = self.read_raw()?;
let dt = self.datatype()?;
Ok(data_read::read_as_u64(&raw, &dt)?)
}
pub fn read_u8(&self) -> Result<Vec<u8>, Error> {
self.read_raw()
}
#[expect(
clippy::cast_possible_wrap,
reason = "read_i8 reinterprets each stored byte as the signed i8 the caller requested"
)]
pub fn read_i8(&self) -> Result<Vec<i8>, Error> {
let raw = self.read_raw()?;
Ok(raw.iter().map(|&b| b as i8).collect())
}
pub fn read_i16(&self) -> Result<Vec<i16>, Error> {
let raw = self.read_raw()?;
let dt = self.datatype()?;
Ok(data_read::read_as_i16(&raw, &dt)?)
}
pub fn read_u16(&self) -> Result<Vec<u16>, Error> {
let raw = self.read_raw()?;
let dt = self.datatype()?;
Ok(data_read::read_as_u16(&raw, &dt)?)
}
pub fn read_u32(&self) -> Result<Vec<u32>, Error> {
let raw = self.read_raw()?;
let dt = self.datatype()?;
Ok(data_read::read_as_u32(&raw, &dt)?)
}
pub fn read_string(&self) -> Result<Vec<String>, Error> {
let dt = self.datatype()?;
if vl_data::is_vlen_string_datatype(&dt) {
self.read_vlen_strings(VlenStringReadOptions::default())
} else {
let raw = self.read_raw()?;
Ok(data_read::read_as_strings(&raw, &dt)?)
}
}
pub fn vlen_string_payload_size(&self) -> Result<u64, Error> {
let datatype = self.datatype()?;
if !vl_data::is_vlen_string_datatype(&datatype) {
return Err(FormatError::TypeMismatch {
expected: "VariableLength string",
actual: "non-VariableLength string",
}
.into());
}
let dataspace = self.dataspace()?;
let raw = self.read_raw()?;
Ok(vl_data::vlen_string_payload_size(
&raw,
dataspace.num_elements(),
self.file.offset_size(),
)?)
}
pub fn read_vlen_strings(&self, options: VlenStringReadOptions) -> Result<Vec<String>, Error> {
let mut strings = Vec::new();
self.visit_vlen_strings(options, |string| strings.push(string.to_owned()))?;
Ok(strings)
}
pub fn visit_vlen_strings<F>(
&self,
options: VlenStringReadOptions,
visitor: F,
) -> Result<(), Error>
where
F: FnMut(&str),
{
let datatype = self.datatype()?;
if !vl_data::is_vlen_string_datatype(&datatype) {
return Err(FormatError::TypeMismatch {
expected: "VariableLength string",
actual: "non-VariableLength string",
}
.into());
}
let dataspace = self.dataspace()?;
if let Some(limit) = options.max_elements()
&& dataspace.num_elements() > limit as u64
{
return Err(FormatError::VariableLengthElementLimitExceeded {
limit,
actual: dataspace.num_elements(),
}
.into());
}
let raw = self.read_raw()?;
self.file.with_source(|source| {
Ok(vl_data::visit_vl_strings_from_source(
source,
&raw,
dataspace.num_elements(),
self.file.offset_size(),
self.file.length_size(),
self.file.addr_offset,
options,
visitor,
)?)
})
}
pub(crate) fn read_vlen_string_bytes(
&self,
options: VlenStringReadOptions,
) -> Result<Vec<vl_data::VlByteObject>, Error> {
let datatype = self.datatype()?;
if !vl_data::is_vlen_string_datatype(&datatype) {
return Err(FormatError::TypeMismatch {
expected: "VariableLength string",
actual: "non-VariableLength string",
}
.into());
}
let dataspace = self.dataspace()?;
if let Some(limit) = options.max_elements()
&& dataspace.num_elements() > limit as u64
{
return Err(FormatError::VariableLengthElementLimitExceeded {
limit,
actual: dataspace.num_elements(),
}
.into());
}
let raw = self.read_raw()?;
self.file.with_source(|source| {
Ok(vl_data::read_vl_byte_objects_from_source(
source,
&raw,
dataspace.num_elements(),
self.file.offset_size(),
self.file.length_size(),
self.file.addr_offset,
1, options,
)?)
})
}
pub(crate) fn read_vlen_sequence_bytes(
&self,
options: VlenStringReadOptions,
) -> Result<(Vec<vl_data::VlByteObject>, usize), Error> {
let datatype = self.datatype()?;
let Datatype::VariableLength { base_type, .. } = &datatype else {
return Err(FormatError::TypeMismatch {
expected: "non-string VariableLength",
actual: "non-VariableLength",
}
.into());
};
if vl_data::is_vlen_string_datatype(&datatype) {
return Err(FormatError::TypeMismatch {
expected: "non-string VariableLength",
actual: "VariableLength string",
}
.into());
}
let element_size = base_type.type_size() as usize;
if element_size == 0 {
return Err(
FormatError::VlDataError("non-string VL base type has zero size".into()).into(),
);
}
let dataspace = self.dataspace()?;
if let Some(limit) = options.max_elements()
&& dataspace.num_elements() > limit as u64
{
return Err(FormatError::VariableLengthElementLimitExceeded {
limit,
actual: dataspace.num_elements(),
}
.into());
}
let raw = self.read_raw()?;
let objects = self.file.with_source(|source| {
vl_data::read_vl_byte_objects_from_source(
source,
&raw,
dataspace.num_elements(),
self.file.offset_size(),
self.file.length_size(),
self.file.addr_offset,
element_size,
options,
)
})?;
Ok((objects, element_size))
}
pub(crate) fn read_embedded_vlen_bytes(
&self,
slots: &[vl_data::EmbeddedVlSlot],
options: VlenStringReadOptions,
) -> Result<vl_data::EmbeddedVlData, Error> {
let stride = self.datatype()?.type_size() as usize;
let dataspace = self.dataspace()?;
let n = dataspace.num_elements();
if let Some(limit) = options.max_elements()
&& n > limit as u64
{
return Err(
FormatError::VariableLengthElementLimitExceeded { limit, actual: n }.into(),
);
}
let raw = if n == 0 { Vec::new() } else { self.read_raw()? };
let n_usize = n.to_usize()?;
let needed = n_usize
.checked_mul(stride)
.ok_or(FormatError::OffsetOverflow {
offset: n,
length: stride as u64,
})?;
if raw.len() < needed {
return Err(FormatError::UnexpectedEof {
expected: needed,
available: raw.len(),
}
.into());
}
let mut offsets = Vec::with_capacity(n_usize * slots.len());
let mut objects = Vec::with_capacity(n_usize * slots.len());
for slot in slots {
let mut dense = Vec::with_capacity(n_usize * VL_REF_SIZE);
for e in 0..n_usize {
let at = e * stride + slot.byte_offset;
dense.extend_from_slice(&raw[at..at + VL_REF_SIZE]);
offsets.push(at);
}
let resolved = self.file.with_source(|source| {
vl_data::read_vl_byte_objects_from_source(
source,
&dense,
n,
self.file.offset_size(),
self.file.length_size(),
self.file.addr_offset,
slot.element_size,
options,
)
})?;
objects.extend(resolved);
}
Ok(vl_data::EmbeddedVlData {
raw,
offsets,
objects,
})
}
pub fn attrs(&self) -> Result<HashMap<String, AttrValue>, Error> {
self.file.attrs_of(&self.header)
}
pub(crate) fn attr_names(&self) -> Result<Vec<String>, Error> {
self.file.attr_message_names_of(&self.header)
}
pub fn datatype(&self) -> Result<Datatype, Error> {
let msg = find_message(&self.header, MessageType::Datatype)?;
let (dt, _) = Datatype::parse(&msg.data)?;
Ok(dt)
}
pub(crate) fn dataspace(&self) -> Result<Dataspace, Error> {
let msg = find_message(&self.header, MessageType::Dataspace)?;
Ok(Dataspace::parse(&msg.data, self.file.length_size())?)
}
pub(crate) fn data_layout(&self) -> Result<DataLayout, Error> {
let msg = find_message(&self.header, MessageType::DataLayout)?;
Ok(DataLayout::parse(
&msg.data,
self.file.offset_size(),
self.file.length_size(),
)?)
}
pub(crate) fn filter_pipeline_parsed(&self) -> Option<FilterPipeline> {
self.header
.messages
.iter()
.find(|m| m.msg_type == MessageType::FilterPipeline)
.and_then(|msg| FilterPipeline::parse(&msg.data).ok())
}
pub(crate) fn raw_chunks(&self) -> Result<Vec<crate::chunked_read::ChunkInfo>, Error> {
let DataLayout::Chunked {
chunk_dimensions,
btree_address,
version,
chunk_index_type,
single_chunk_filtered_size,
single_chunk_filter_mask,
} = self.data_layout()?
else {
return Err(Error::Format(crate::error::FormatError::ChunkedReadError(
"chunk enumeration requires a chunked dataset".into(),
)));
};
let Some(addr) = btree_address else {
return Ok(Vec::new());
};
let dataspace = self.dataspace()?;
let elem_size = self.datatype()?.type_size() as usize;
let base = self.file.addr_offset;
self.file.with_source(|source| {
if base == 0 {
return Ok(crate::chunked_read::collect_chunks_for_layout_from_source(
source,
version,
chunk_index_type,
addr,
single_chunk_filtered_size,
single_chunk_filter_mask,
&chunk_dimensions,
&dataspace,
elem_size,
self.file.offset_size(),
self.file.length_size(),
)?);
}
let framed = BaseOffsetSource {
inner: source,
base,
};
let mut chunks = crate::chunked_read::collect_chunks_for_layout_from_source(
&framed,
version,
chunk_index_type,
addr,
single_chunk_filtered_size,
single_chunk_filter_mask,
&chunk_dimensions,
&dataspace,
elem_size,
self.file.offset_size(),
self.file.length_size(),
)?;
for c in &mut chunks {
c.address = c.address.checked_add(base).ok_or(
crate::error::FormatError::OffsetOverflow {
offset: c.address,
length: 0,
},
)?;
}
Ok(chunks)
})
}
pub(crate) fn filter_pipeline_message_bytes(&self) -> Option<Vec<u8>> {
self.header
.messages
.iter()
.find(|m| m.msg_type == MessageType::FilterPipeline)
.map(|msg| msg.data.clone())
}
pub fn read_raw(&self) -> Result<Vec<u8>, Error> {
let dt = self.datatype()?;
let ds = self.dataspace()?;
let dl = self.data_layout()?;
let pipeline = self.filter_pipeline_parsed();
Ok(self
.file
.read_dataset_raw(&dl, &ds, &dt, pipeline.as_ref(), &self.chunk_cache)?)
}
pub fn read_raw_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<u8>, Error> {
let dt = self.datatype()?;
let ds = self.dataspace()?;
let dl = self.data_layout()?;
let n0 = ds.dimensions.first().copied().unwrap_or(1);
let start = start_row.min(n0);
let count = num_rows.min(n0 - start);
if start == 0 && count == n0 {
let pipeline = self.filter_pipeline_parsed();
return Ok(self.file.read_dataset_raw(
&dl,
&ds,
&dt,
pipeline.as_ref(),
&self.chunk_cache,
)?);
}
Ok(self.file.read_dataset_raw_rows(
&dl,
&ds,
&dt,
self.filter_pipeline_parsed().as_ref(),
&self.chunk_cache,
start,
count,
)?)
}
pub fn read_f64_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<f64>, Error> {
let raw = self.read_raw_rows(start_row, num_rows)?;
Ok(data_read::read_as_f64(&raw, &self.datatype()?)?)
}
pub fn read_f32_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<f32>, Error> {
let raw = self.read_raw_rows(start_row, num_rows)?;
Ok(data_read::read_as_f32(&raw, &self.datatype()?)?)
}
#[expect(
clippy::cast_possible_wrap,
reason = "read_i8 reinterprets each stored byte as the signed i8 the caller requested"
)]
pub fn read_i8_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<i8>, Error> {
let raw = self.read_raw_rows(start_row, num_rows)?;
Ok(raw.iter().map(|&b| b as i8).collect())
}
pub fn read_i16_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<i16>, Error> {
let raw = self.read_raw_rows(start_row, num_rows)?;
Ok(data_read::read_as_i16(&raw, &self.datatype()?)?)
}
pub fn read_i32_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<i32>, Error> {
let raw = self.read_raw_rows(start_row, num_rows)?;
Ok(data_read::read_as_i32(&raw, &self.datatype()?)?)
}
pub fn read_i64_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<i64>, Error> {
let raw = self.read_raw_rows(start_row, num_rows)?;
Ok(data_read::read_as_i64(&raw, &self.datatype()?)?)
}
pub fn read_u8_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<u8>, Error> {
self.read_raw_rows(start_row, num_rows)
}
pub fn read_u16_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<u16>, Error> {
let raw = self.read_raw_rows(start_row, num_rows)?;
Ok(data_read::read_as_u16(&raw, &self.datatype()?)?)
}
pub fn read_u32_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<u32>, Error> {
let raw = self.read_raw_rows(start_row, num_rows)?;
Ok(data_read::read_as_u32(&raw, &self.datatype()?)?)
}
pub fn read_u64_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<u64>, Error> {
let raw = self.read_raw_rows(start_row, num_rows)?;
Ok(data_read::read_as_u64(&raw, &self.datatype()?)?)
}
pub fn read_string_rows(&self, start_row: u64, num_rows: u64) -> Result<Vec<String>, Error> {
let dt = self.datatype()?;
if vl_data::is_vlen_string_datatype(&dt) {
let raw = self.read_raw_rows(start_row, num_rows)?;
let ref_size = 4 + self.file.offset_size() as usize + 4;
let num_elements = (raw.len() / ref_size) as u64;
let mut strings = Vec::new();
self.file.with_source(|source| -> Result<(), Error> {
Ok(vl_data::visit_vl_strings_from_source(
source,
&raw,
num_elements,
self.file.offset_size(),
self.file.length_size(),
self.file.addr_offset,
VlenStringReadOptions::default(),
|string| strings.push(String::from(string)),
)?)
})?;
return Ok(strings);
}
let raw = self.read_raw_rows(start_row, num_rows)?;
Ok(data_read::read_as_strings(&raw, &dt)?)
}
pub fn dereference(&self) -> Result<Vec<Object>, Error> {
let dt = self.datatype()?;
if !matches!(
dt,
Datatype::Reference {
ref_type: ReferenceType::Object,
..
}
) {
return Err(FormatError::TypeMismatch {
expected: "object reference",
actual: "non-reference datatype",
}
.into());
}
let elem_size = dt.type_size().to_usize()?;
if elem_size < 8 {
return Err(FormatError::TypeMismatch {
expected: "8-byte object reference",
actual: "object reference narrower than 8 bytes",
}
.into());
}
let raw = self.read_raw()?;
if raw.is_empty() {
return Ok(Vec::new());
}
if !raw.len().is_multiple_of(elem_size) {
return Err(FormatError::DataSizeMismatch {
expected: elem_size,
actual: raw.len(),
}
.into());
}
let mut out = Vec::with_capacity(raw.len() / elem_size);
for chunk in raw.chunks_exact(elem_size) {
let addr = u64::from_le_bytes(chunk[..8].try_into().expect("chunk has >= 8 bytes"));
out.push(FileInner::object_at_relative(&self.file, addr)?);
}
Ok(out)
}
pub fn read_compound<T: CompoundType>(&self) -> Result<Vec<T>, Error> {
let datatype = self.datatype()?;
let element_size = datatype.type_size().to_usize()?;
if !matches!(datatype, Datatype::Compound { .. }) {
return Err(FormatError::TypeMismatch {
expected: "Compound",
actual: "non-Compound",
}
.into());
}
let raw = self.read_raw()?;
if element_size == 0 || !raw.len().is_multiple_of(element_size) {
return Err(FormatError::DataSizeMismatch {
expected: element_size,
actual: raw.len(),
}
.into());
}
raw.chunks_exact(element_size)
.map(|bytes| T::decode(&datatype, bytes).map_err(Error::from))
.collect()
}
#[cfg(feature = "provenance")]
pub fn verify_provenance(&self) -> Result<crate::provenance::VerifyResult, Error> {
use crate::provenance::{ATTR_SHA256, VerifyResult, sha256_hex};
let attrs = self.attrs()?;
let stored = match attrs.get(ATTR_SHA256) {
Some(AttrValue::String(s) | AttrValue::AsciiString(s)) => {
s.trim_end_matches('\0').to_string()
}
_ => return Ok(VerifyResult::NoHash),
};
let computed = sha256_hex(&self.read_raw()?);
if computed == stored {
Ok(VerifyResult::Ok)
} else {
Ok(VerifyResult::Mismatch { stored, computed })
}
}
}
fn find_message(
header: &ObjectHeader,
msg_type: MessageType,
) -> Result<&crate::object_header::HeaderMessage, Error> {
header
.messages
.iter()
.find(|m| m.msg_type == msg_type)
.ok_or(Error::MissingMessage(msg_type))
}
fn normalize_path(path: &str) -> String {
path.trim_matches('/').to_string()
}
fn has_message(header: &ObjectHeader, msg_type: MessageType) -> bool {
header.messages.iter().any(|m| m.msg_type == msg_type)
}
fn is_group(header: &ObjectHeader) -> bool {
header.messages.iter().any(|m| {
m.msg_type == MessageType::LinkInfo
|| m.msg_type == MessageType::Link
|| m.msg_type == MessageType::SymbolTable
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::FileBuilder;
fn chunked_file_bytes() -> Vec<u8> {
let data: Vec<i32> = (0..256).collect();
let mut b = FileBuilder::new();
b.create_dataset("chunked")
.with_i32_data(&data)
.with_shape(&[256])
.with_chunks(&[32]);
b.finish().unwrap()
}
#[test]
fn enabled_override_populates_live_cache_over_disabled_file_default() {
let file = File::from_bytes_with_options(
chunked_file_bytes(),
FileAccessProperties::new().with_chunk_cache(ChunkCacheConfig::disabled()),
)
.unwrap();
let ds = file
.dataset_with_options(
"chunked",
DatasetAccessProperties::new().with_chunk_cache(ChunkCacheConfig::new()),
)
.unwrap();
assert_eq!(ds.read_i32().unwrap(), (0..256).collect::<Vec<i32>>());
assert!(ds.chunk_cache_stats().index_loaded());
assert!(ds.chunk_cache_stats().cached_chunks() > 0);
}
#[test]
fn disabled_override_suppresses_live_cache_over_enabled_file_default() {
let file = File::from_bytes_with_options(
chunked_file_bytes(),
FileAccessProperties::new().with_chunk_cache(ChunkCacheConfig::new()),
)
.unwrap();
let ds = file
.dataset_with_options(
"chunked",
DatasetAccessProperties::new().with_chunk_cache(ChunkCacheConfig::disabled()),
)
.unwrap();
assert_eq!(ds.read_i32().unwrap(), (0..256).collect::<Vec<i32>>());
assert!(!ds.chunk_cache_stats().index_loaded());
assert_eq!(ds.chunk_cache_stats().cached_chunks(), 0);
}
#[test]
fn group_child_address_base_overflow_is_rejected() {
const UB: u64 = 512;
let mut b = FileBuilder::new();
b.with_userblock(UB);
let mut child = b.create_group("child");
child.create_dataset("inner").with_i32_data(&[1, 2, 3]);
b.add_group(child.finish());
let mut bytes = b.finish().unwrap();
let file = File::from_bytes(bytes.clone()).unwrap();
assert_eq!(file.root().groups().unwrap(), vec!["child".to_string()]);
let stored = file.root().group("child").unwrap().address - UB;
let needle = stored.to_le_bytes();
let matches: Vec<usize> = bytes
.windows(8)
.enumerate()
.filter(|(_, w)| *w == needle)
.map(|(i, _)| i)
.collect();
assert_eq!(
matches.len(),
1,
"stored child address {stored:#x} was not uniquely locatable: {matches:?}"
);
bytes[matches[0]..matches[0] + 8].copy_from_slice(&u64::MAX.to_le_bytes());
#[cfg(feature = "checksum")]
{
let root_addr = file.root().address as usize;
assert_eq!(&bytes[root_addr..root_addr + 4], b"OHDR");
let flags = bytes[root_addr + 5];
let mut pos = root_addr + 6;
if flags & 0x20 != 0 {
pos += 16;
}
if flags & 0x10 != 0 {
pos += 4;
}
let width = 1usize << (flags & 0x03);
let chunk0 = (0..width).fold(0usize, |acc, i| {
acc | ((bytes[pos + i] as usize) << (8 * i))
});
pos += width;
let chunk0_end = pos + chunk0;
assert!(
matches[0] < chunk0_end,
"patched link address is outside the root header's chunk-0"
);
let cs = crate::checksum::jenkins_lookup3(&bytes[root_addr..chunk0_end]);
bytes[chunk0_end..chunk0_end + 4].copy_from_slice(&cs.to_le_bytes());
}
let file = File::from_bytes(bytes).unwrap();
match file.root().groups() {
Err(Error::Format(FormatError::OffsetOverflow { offset, length })) => {
assert_eq!(offset, u64::MAX);
assert_eq!(length, UB);
}
other => panic!("expected group-child address overflow, got {other:?}"),
}
}
#[test]
fn read_rows_framed_zero_row_window_is_ok_even_when_unallocated() {
let dl = DataLayout::Contiguous {
address: None,
size: 0,
};
let ds = Dataspace {
space_type: crate::dataspace::DataspaceType::Simple,
rank: 1,
dimensions: vec![0],
max_dimensions: None,
};
let dt = Datatype::FixedPoint {
size: 8,
byte_order: crate::datatype::DatatypeByteOrder::LittleEndian,
signed: false,
bit_offset: 0,
bit_precision: 64,
};
let cache = ChunkCache::new();
let out = read_rows_framed(
&BytesSource::new(b""),
&dl,
&ds,
&dt,
None,
8,
8,
&cache,
0,
0,
8,
)
.expect("a zero-row window must be Ok(empty), not NoDataAllocated");
assert!(out.is_empty());
let virtual_dl = DataLayout::Virtual { version: 4 };
let err = read_rows_framed(
&BytesSource::new(b""),
&virtual_dl,
&ds,
&dt,
None,
8,
8,
&cache,
0,
0,
8,
)
.expect_err("a virtual layout must error even for a zero-row window");
assert!(
matches!(err, FormatError::UnsupportedVirtualLayout),
"expected UnsupportedVirtualLayout, got {err:?}"
);
}
}