From 9a99dce278712a04d440ad0c15e9cead483b6532 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Tue, 18 Nov 2025 08:09:11 +0000 Subject: [PATCH 01/24] One read for b-tree nodes --- pyfive/btree.py | 29 ++++++++++++++++++++++------- 1 file changed, 22 insertions(+), 7 deletions(-) diff --git a/pyfive/btree.py b/pyfive/btree.py index 88d53f50..caaa311f 100644 --- a/pyfive/btree.py +++ b/pyfive/btree.py @@ -134,14 +134,29 @@ def __init__(self, fh, offset, dims): def _read_node(self, offset, node_level): """ Return a single node in the b-tree located at a give offset. """ node = self._read_node_header(offset, node_level) + keys = [] addresses = [] - for _ in range(node['entries_used']): - chunk_size, filter_mask = struct.unpack(' Date: Wed, 25 Mar 2026 13:52:18 +0000 Subject: [PATCH 02/24] First cut at adding some parallelism in pyfive --- pyfive/h5d.py | 155 ++++++++++++++++++++++++++++++++++++++----- pyfive/high_level.py | 4 +- 2 files changed, 139 insertions(+), 20 deletions(-) diff --git a/pyfive/h5d.py b/pyfive/h5d.py index 2d70b3ae..7a6ee0ad 100644 --- a/pyfive/h5d.py +++ b/pyfive/h5d.py @@ -14,9 +14,11 @@ from io import UnsupportedOperation from time import time +import os import struct import logging from importlib.metadata import version +from concurrent.futures import ThreadPoolExecutor logger = logging.getLogger(__name__) @@ -24,7 +26,140 @@ ChunkIndex = namedtuple("ChunkIndex", "chunk_address chunk_dims") -class DatasetID: + +class ChunkRead: + """ + Mixin providing parallel and bulk chunk-reading strategies. + + ``DatasetID`` inherits from this class so that the hot path in + ``_get_selection_via_chunks`` can dispatch to the best available I/O strategy: + + * **Case A - fsspec ``cat_ranges``**: a single bulk request issued to the + filesystem; ideal for remote stores (S3, GCS, https, …). + * **Case B - ``os.pread`` thread pool**: parallel POSIX reads sharing a + single file descriptor without seek contention. + * **Case C - serial fallback**: safe for in-memory buffers and any + custom file-like wrapper. + """ + + # ------------------------------------------------------------------ # + # Shared helpers # + # ------------------------------------------------------------------ # + + def _get_required_chunks(self, indexer): + """Walk *indexer* and return a list of + ``(chunk_coords, chunk_selection, out_selection, storeinfo)`` + tuples for every chunk needed to satisfy the selection. + """ + result = [] + for chunk_coords, chunk_selection, out_selection in indexer: + chunk_coords = tuple(map(mul, chunk_coords, self.chunks)) + storeinfo = self._index[chunk_coords] + result.append((chunk_coords, chunk_selection, out_selection, storeinfo)) + return result + + def _decode_chunk(self, chunk_buffer, filter_mask, dtype): + """Apply the filter pipeline (if any) and return a shaped ndarray.""" + if self.filter_pipeline is not None: + chunk_buffer = BTreeV1RawDataChunks._filter_chunk( + chunk_buffer, + filter_mask, + self.filter_pipeline, + self.dtype.itemsize, + ) + return np.frombuffer(chunk_buffer, dtype=dtype).reshape( + self.chunks, order=self._order + ) + + def _select_chunks(self, indexer, out, dtype): + """Collect required chunks and dispatch I/O to the best strategy. + Called by ``_get_selection_via_chunks`` in place of the serial loop. + """ + chunks = self._get_required_chunks(indexer) + if not chunks: + return + + # Case A: fsspec – bulk parallel fetch via cat_ranges + if not self.posix: + fh = self._fh + if hasattr(fh, "fs") and hasattr(fh.fs, "cat_ranges"): + self._read_bulk_fsspec(fh, chunks, out, dtype) + return + + # Case B: POSIX – thread-parallel reads via os.pread + if self.posix and hasattr(os, "pread"): + self._read_parallel_threads(chunks, out, dtype) + return + + # Case C: serial fallback + self._read_serial(chunks, out, dtype) + + # ------------------------------------------------------------------ # + # Strategy implementations # + # ------------------------------------------------------------------ # + + def _read_serial(self, chunks, out, dtype): + """Read one chunk at a time (safe for any file-like object).""" + fh = self._fh + for _coords, chunk_sel, out_sel, storeinfo in chunks: + fh.seek(storeinfo.byte_offset) + chunk_buffer = fh.read(storeinfo.size) + out[out_sel] = self._decode_chunk( + chunk_buffer, storeinfo.filter_mask, dtype + )[chunk_sel] + if self.posix: + fh.close() + + def _read_parallel_threads(self, chunks, out, dtype): + """Thread-parallel read via ``os.pread``. + + ``os.pread`` does not advance the file-position pointer, so all + worker threads share a single open file descriptor safely. + """ + fh = open(self._filename, "rb") + fd = fh.fileno() + + def _read_one(item): + _coords, chunk_sel, out_sel, storeinfo = item + return ( + chunk_sel, + out_sel, + storeinfo.filter_mask, + os.pread(fd, storeinfo.size, storeinfo.byte_offset), + ) + + try: + with ThreadPoolExecutor() as executor: + results = list(executor.map(_read_one, chunks)) + finally: + fh.close() + + for chunk_sel, out_sel, filter_mask, chunk_buffer in results: + out[out_sel] = self._decode_chunk(chunk_buffer, filter_mask, dtype)[ + chunk_sel + ] + + def _read_bulk_fsspec(self, fh, chunks, out, dtype): + """Bulk read via ``fsspec`` ``cat_ranges``. + + Issues a single pipelined request for all required byte-ranges, + which on object stores typically translates to a small number of + HTTP range requests rather than one round-trip per chunk. + """ + path = fh.path + starts = [si.byte_offset for _, _, _, si in chunks] + stops = [si.byte_offset + si.size for _, _, _, si in chunks] + buffers = fh.fs.cat_ranges([path] * len(chunks), starts, stops) + + for (_coords, chunk_sel, out_sel, storeinfo), chunk_buffer in zip( + chunks, buffers + ): + out[out_sel] = self._decode_chunk( + chunk_buffer, storeinfo.filter_mask, dtype + )[chunk_sel] + + +class DatasetID(ChunkRead): """ Implements an "HDF5 dataset identifier", which despite the name, actually represents the data of a dataset in a file, and not an identifier. It includes all @@ -679,23 +814,7 @@ def _get_selection_via_chunks(self, args): fh.close() else: - for chunk_coords, chunk_selection, out_selection in indexer: - # map from chunk coordinate space to array space which - # is how hdf5 keeps the index - chunk_coords = tuple(map(mul, chunk_coords, self.chunks)) - filter_mask, chunk_buffer = self.read_direct_chunk(chunk_coords) - if self.filter_pipeline is not None: - # we are only using the class method here, future - # filter pipelines may need their own function - chunk_buffer = BTreeV1RawDataChunks._filter_chunk( - chunk_buffer, - filter_mask, - self.filter_pipeline, - self.dtype.itemsize, - ) - chunk_data = np.frombuffer(chunk_buffer, dtype=dtype) - chunk_data = chunk_data.reshape(self.chunks, order=self._order) - out[out_selection] = chunk_data[chunk_selection] + self._select_chunks(indexer, out, dtype) if isinstance(self._ptype, P5ReferenceType): to_reference = np.vectorize(Reference) diff --git a/pyfive/high_level.py b/pyfive/high_level.py index 242bb2e7..fe650b55 100644 --- a/pyfive/high_level.py +++ b/pyfive/high_level.py @@ -281,12 +281,12 @@ def __init__( # Already wrapped self._fh = fh elif type(fh).__name__ == "S3File" or hasattr(fh, "fs"): - # S3 file handle - wrap with buffering + # fsspec file handle - wrap with buffering # We check for the S3File type by name to avoid a hard dependency on s3fs, # but also check for an 'fs' attribute which is common in s3fs file-like objects. # This may yet be too broad, but it is unlikely to cause issues for non-S3 files. logger.info( - "[pyfive] Detected S3 file, enabling metadata buffering (%d MB)", + "[pyfive] Detected remote file, enabling metadata buffering (%d MB)", metadata_buffer_size, ) self._fh = MetadataBufferingWrapper(fh, buffer_size=metadata_buffer_size) From cca72e52d5bdf19a867f08c63aa756718a3cd389 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Wed, 25 Mar 2026 14:37:29 +0000 Subject: [PATCH 03/24] controlling and logging parallelism --- pyfive/h5d.py | 63 ++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 52 insertions(+), 11 deletions(-) diff --git a/pyfive/h5d.py b/pyfive/h5d.py index 7a6ee0ad..b19c8c3a 100644 --- a/pyfive/h5d.py +++ b/pyfive/h5d.py @@ -46,8 +46,34 @@ class ChunkRead: # Shared helpers # # ------------------------------------------------------------------ # + def set_parallelism(self, thread_count=0, cat_range_allowed=False): + """ + Configure experimental chunk-read parallelism. + + ``thread_count`` controls POSIX threaded reads via ``os.pread``: + - ``0`` disables threaded reads + - ``>0`` enables threaded reads with that many workers + + ``cat_range_allowed`` enables fsspec bulk reads via ``cat_ranges`` + for compatible non-posix file handles. + + This is a ``pyfive`` API extension, and is opt-in by default as it may not be suitable for all use cases. + It is recommended to enable it when working with remote files, but it may not be suitable for local files. + + """ + if thread_count is None: + thread_count = 0 + thread_count = int(thread_count) + if thread_count < 0: + raise ValueError("thread_count must be >= 0") + + self._thread_count = thread_count + self._cat_range_allowed = bool(cat_range_allowed) + + def _get_required_chunks(self, indexer): - """Walk *indexer* and return a list of + """ + Walk *indexer* and return a list of ``(chunk_coords, chunk_selection, out_selection, storeinfo)`` tuples for every chunk needed to satisfy the selection. """ @@ -59,7 +85,9 @@ def _get_required_chunks(self, indexer): return result def _decode_chunk(self, chunk_buffer, filter_mask, dtype): - """Apply the filter pipeline (if any) and return a shaped ndarray.""" + """ + Apply the filter pipeline (if any) and return a shaped ndarray + """ if self.filter_pipeline is not None: chunk_buffer = BTreeV1RawDataChunks._filter_chunk( chunk_buffer, @@ -72,26 +100,33 @@ def _decode_chunk(self, chunk_buffer, filter_mask, dtype): ) def _select_chunks(self, indexer, out, dtype): - """Collect required chunks and dispatch I/O to the best strategy. + """ + Collect required chunks and dispatch I/O to the best strategy. Called by ``_get_selection_via_chunks`` in place of the serial loop. """ chunks = self._get_required_chunks(indexer) if not chunks: return - # Case A: fsspec – bulk parallel fetch via cat_ranges - if not self.posix: + # Case A: fsspec - bulk parallel fetch via cat_ranges + if not self.posix and self._cat_range_allowed: fh = self._fh if hasattr(fh, "fs") and hasattr(fh.fs, "cat_ranges"): + logger.debug("[pyfive] chunk read strategy: fsspec_cat_ranges") self._read_bulk_fsspec(fh, chunks, out, dtype) return - # Case B: POSIX – thread-parallel reads via os.pread - if self.posix and hasattr(os, "pread"): + # Case B: POSIX - thread-parallel reads via os.pread + if self.posix and hasattr(os, "pread") and self._thread_count != 0: + logger.debug( + "[pyfive] chunk read strategy: posix_pread_threads workers=%s", + self._thread_count, + ) self._read_parallel_threads(chunks, out, dtype) return # Case C: serial fallback + logger.debug("[pyfive] chunk read strategy: serial") self._read_serial(chunks, out, dtype) # ------------------------------------------------------------------ # @@ -99,7 +134,9 @@ def _select_chunks(self, indexer, out, dtype): # ------------------------------------------------------------------ # def _read_serial(self, chunks, out, dtype): - """Read one chunk at a time (safe for any file-like object).""" + """ + Read one chunk at a time (safe for any file-like object). + """ fh = self._fh for _coords, chunk_sel, out_sel, storeinfo in chunks: fh.seek(storeinfo.byte_offset) @@ -111,7 +148,8 @@ def _read_serial(self, chunks, out, dtype): fh.close() def _read_parallel_threads(self, chunks, out, dtype): - """Thread-parallel read via ``os.pread``. + """ + Thread-parallel read via ``os.pread``. ``os.pread`` does not advance the file-position pointer, so all worker threads share a single open file descriptor safely. @@ -129,7 +167,7 @@ def _read_one(item): ) try: - with ThreadPoolExecutor() as executor: + with ThreadPoolExecutor(max_workers=self._thread_count) as executor: results = list(executor.map(_read_one, chunks)) finally: fh.close() @@ -140,7 +178,8 @@ def _read_one(item): ] def _read_bulk_fsspec(self, fh, chunks, out, dtype): - """Bulk read via ``fsspec`` ``cat_ranges``. + """ + Bulk read via ``fsspec`` ``cat_ranges``. Issues a single pipelined request for all required byte-ranges, which on object stores typically translates to a small number of @@ -235,6 +274,8 @@ def __init__( self.shape = dataobject.shape self.rank = len(self.shape) self.chunks = dataobject.chunks + # Experimental chunk-read settings are opt-in by default. + self.set_parallelism() # experimental code. We need to find out whether or not this # is unnecessary duplication. At the moment it seems best for From 0fe1e802f3014f61349347a25c6bc7a5d5150c66 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Tue, 31 Mar 2026 12:21:10 +0100 Subject: [PATCH 04/24] Making parallelism actually work --- pyfive/h5d.py | 23 +++++++++++++++-------- pyfive/utilities.py | 11 +++++++++++ 2 files changed, 26 insertions(+), 8 deletions(-) diff --git a/pyfive/h5d.py b/pyfive/h5d.py index b19c8c3a..00a55ce4 100644 --- a/pyfive/h5d.py +++ b/pyfive/h5d.py @@ -69,6 +69,7 @@ def set_parallelism(self, thread_count=0, cat_range_allowed=False): self._thread_count = thread_count self._cat_range_allowed = bool(cat_range_allowed) + logger.info('Parallelism set with thread_count=%d and cat_range_allowed=%s', self._thread_count, self._cat_range_allowed) def _get_required_chunks(self, indexer): @@ -111,22 +112,24 @@ def _select_chunks(self, indexer, out, dtype): # Case A: fsspec - bulk parallel fetch via cat_ranges if not self.posix and self._cat_range_allowed: fh = self._fh - if hasattr(fh, "fs") and hasattr(fh.fs, "cat_ranges"): - logger.debug("[pyfive] chunk read strategy: fsspec_cat_ranges") + actual_fh = getattr(fh, "fh", fh) # support wrapped file-like objects + if hasattr(actual_fh, "fs") and hasattr(actual_fh.fs, "cat_ranges"): + logger.info(f"[pyfive] chunk read strategy: fsspec_cat_ranges ({len(chunks)} chunks)") self._read_bulk_fsspec(fh, chunks, out, dtype) return # Case B: POSIX - thread-parallel reads via os.pread if self.posix and hasattr(os, "pread") and self._thread_count != 0: - logger.debug( - "[pyfive] chunk read strategy: posix_pread_threads workers=%s", + logger.info( + "[pyfive] chunk read strategy: posix_pread_threads workers=%s (%d chunks)", self._thread_count, + len(chunks), ) self._read_parallel_threads(chunks, out, dtype) return # Case C: serial fallback - logger.debug("[pyfive] chunk read strategy: serial") + logger.info("[pyfive] chunk read strategy: serial (%d chunks)", len(chunks)) self._read_serial(chunks, out, dtype) # ------------------------------------------------------------------ # @@ -172,6 +175,8 @@ def _read_one(item): finally: fh.close() + logger.info('pyfive thread pool read completed using %d threads', self._thread_count) + for chunk_sel, out_sel, filter_mask, chunk_buffer in results: out[out_sel] = self._decode_chunk(chunk_buffer, filter_mask, dtype)[ chunk_sel @@ -183,12 +188,14 @@ def _read_bulk_fsspec(self, fh, chunks, out, dtype): Issues a single pipelined request for all required byte-ranges, which on object stores typically translates to a small number of - HTTP range requests rather than one round-trip per chunk. + HTTP range requests rather than one round-trip per chunk + (reaching through the MetadataBufferingWrapper). """ - path = fh.path + actual_fh = getattr(fh, "fh", fh) # support wrapped file-like objects + path = actual_fh.path starts = [si.byte_offset for _, _, _, si in chunks] stops = [si.byte_offset + si.size for _, _, _, si in chunks] - buffers = fh.fs.cat_ranges([path] * len(chunks), starts, stops) + buffers = actual_fh.fs.cat_ranges([path] * len(chunks), starts, stops) for (_coords, chunk_sel, out_sel, storeinfo), chunk_buffer in zip( chunks, buffers diff --git a/pyfive/utilities.py b/pyfive/utilities.py index 3ffeb14e..7471be1f 100644 --- a/pyfive/utilities.py +++ b/pyfive/utilities.py @@ -77,6 +77,17 @@ def closed(self) -> bool: """Return whether file is closed.""" return self._is_closed + @property + def fs(self): + return getattr(self.fh, "fs", None) + + @property + def path(self): + return getattr(self.fh, "path", None) + + def __getattr__(self,name:str): + return getattr(self.fh, name) + def _ensure_buffer(self): """Eagerly read metadata buffer on first access.""" if self.buffer is None: From 00e2f885fca2c4a14bd01a9fcc7623dc01d7b9e4 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Wed, 1 Apr 2026 13:34:42 +0100 Subject: [PATCH 05/24] Optimize B-tree node reads and parallelize leaf fetches --- pyfive/btree.py | 90 ++++++++++++++++++++++ tests/test_btree_parallel.py | 141 +++++++++++++++++++++++++++++++++++ 2 files changed, 231 insertions(+) create mode 100644 tests/test_btree_parallel.py diff --git a/pyfive/btree.py b/pyfive/btree.py index 72718109..2aeadf7b 100644 --- a/pyfive/btree.py +++ b/pyfive/btree.py @@ -55,6 +55,16 @@ def _read_node(self, offset, node_level): self.last_offset = max(offset, self.last_offset) return node + @staticmethod + def _get_cat_ranges_fs(fh): + """Return (fs, path) when cat_ranges is supported, else (None, None).""" + actual_fh = getattr(fh, "fh", fh) + fs = getattr(actual_fh, "fs", None) + path = getattr(actual_fh, "path", None) + if fs is not None and path is not None and hasattr(fs, "cat_ranges"): + return fs, path + return None, None + def _read_node_header(self, offset): """Return a single node header in the b-tree located at a give offset.""" raise NotImplementedError @@ -131,6 +141,42 @@ def __init__(self, fh, offset, dims): self.dims = dims super().__init__(fh, offset) + def _read_children(self): + """Read child nodes; fetch leaf-level nodes in bulk when possible.""" + fs, path = self._get_cat_ranges_fs(self.fh) + if fs is None: + super()._read_children() + return + + # Leaf-level root node: already read in _read_root_node. + if self.depth == 0: + return + + # Traverse internal levels sequentially; each level depends on the prior. + for node_level in range(self.depth, 0, -1): + for parent_node in self.all_nodes[node_level]: + for child_addr in parent_node["addresses"]: + if node_level - 1 > 0: + child_node = self._read_node(child_addr, node_level - 1) + self._add_node(child_node) + + # Collect leaf addresses from the lowest internal level. + leaf_addresses = [] + for node in self.all_nodes.get(1, []): + leaf_addresses.extend(node["addresses"]) + + if not leaf_addresses: + return + + leaf_size = self._estimate_leaf_node_size(leaf_addresses[0]) + starts = list(leaf_addresses) + stops = [addr + leaf_size for addr in starts] + raw_leaves = fs.cat_ranges([path] * len(starts), starts, stops) + + for addr, raw in zip(starts, raw_leaves): + node = self._parse_leaf_from_buffer(raw, addr) + self._add_node(node) + def _read_node(self, offset, node_level): """Return a single node in the b-tree located at a give offset.""" node = self._read_node_header(offset, node_level) @@ -171,6 +217,50 @@ def _read_node(self, offset, node_level): self.last_offset = max(offset, self.last_offset) return node + def _estimate_leaf_node_size(self, first_leaf_addr): + """Estimate leaf-node byte size using the first leaf header.""" + node = self._read_node_header(first_leaf_addr, 0) + header_size = struct.calcsize("<" + "".join(self.B_LINK_NODE.values())) + entry_size = 8 + (8 * self.dims) + 8 + return header_size + (node["entries_used"] * entry_size) + + def _parse_leaf_from_buffer(self, raw, addr): + """Parse a level-0 raw-data chunk node from a bytes buffer.""" + node = _unpack_struct_from(self.B_LINK_NODE, raw) + assert node["signature"] == b"TREE" + assert node["node_type"] == self.NODE_TYPE + assert node["node_level"] == 0 + + keys = [] + addresses = [] + entry_offset_fmt = "<" + "Q" * self.dims + entry_offset_size = struct.calcsize(entry_offset_fmt) + cursor = struct.calcsize("<" + "".join(self.B_LINK_NODE.values())) + + for _ in range(node["entries_used"]): + chunk_size, filter_mask = struct.unpack_from(" Date: Wed, 1 Apr 2026 14:28:42 +0100 Subject: [PATCH 06/24] pyfive --- tests/test_btree_parallel.py | 130 +++++++++++++++++++---------------- 1 file changed, 70 insertions(+), 60 deletions(-) diff --git a/tests/test_btree_parallel.py b/tests/test_btree_parallel.py index 0f5e88f1..7021bcca 100644 --- a/tests/test_btree_parallel.py +++ b/tests/test_btree_parallel.py @@ -1,7 +1,8 @@ import io import struct -from pyfive.btree import AbstractBTree, BTreeV1RawDataChunks +from pyfive.btree import BTreeV1, BTreeV1RawDataChunks +from pyfive.h5d import DatasetID def _build_leaf_node_bytes(*, dims, entries): @@ -19,6 +20,7 @@ def __init__(self, fs, path): self.fh = self self.fs = fs self.path = path + self.closed = False class _DummyFS: @@ -31,57 +33,21 @@ def cat_ranges(self, paths, starts, stops): return [self.payload_by_start[s] for s in starts] -def test_get_cat_ranges_fs_support_detection(): - fs = _DummyFS({}) - wrapped = _WrappedFH(fs, "bucket/file.h5") - out_fs, out_path = AbstractBTree._get_cat_ranges_fs(wrapped) - assert out_fs is fs - assert out_path == "bucket/file.h5" - - class NoCatRangesFS: - pass - - no_support = _WrappedFH(NoCatRangesFS(), "bucket/file.h5") - out_fs, out_path = AbstractBTree._get_cat_ranges_fs(no_support) - assert out_fs is None - assert out_path is None - - -def test_raw_chunk_tree_reads_leaf_nodes_via_cat_ranges(monkeypatch): +def test_read_children_with_fetch_fn_fetches_leaves_once(monkeypatch): leaf_raw = { - 1000: _build_leaf_node_bytes( - dims=1, - entries=[(16, 0, (0,), 7000)], - ), - 1001: _build_leaf_node_bytes( - dims=1, - entries=[(16, 0, (2,), 7001)], - ), - 2000: _build_leaf_node_bytes( - dims=1, - entries=[(16, 0, (4,), 7002)], - ), + 1000: _build_leaf_node_bytes(dims=1, entries=[(16, 0, (0,), 7000)]), + 1001: _build_leaf_node_bytes(dims=1, entries=[(16, 0, (2,), 7001)]), + 2000: _build_leaf_node_bytes(dims=1, entries=[(16, 0, (4,), 7002)]), } - fs = _DummyFS(leaf_raw) tree = BTreeV1RawDataChunks.__new__(BTreeV1RawDataChunks) - tree.fh = _WrappedFH(fs, "bucket/file.h5") + tree.fh = io.BytesIO(b"") tree.dims = 1 tree.depth = 2 tree.last_offset = 0 - tree.all_nodes = { - 2: [ - { - "node_level": 2, - "addresses": [100, 200], - } - ] - } + tree.all_nodes = {2: [{"node_level": 2, "addresses": [100, 200]}]} - children = { - 100: [1000, 1001], - 200: [2000], - } + children = {100: [1000, 1001], 200: [2000]} def fake_read_node(offset, node_level): assert node_level == 1 @@ -92,42 +58,40 @@ def fake_read_node(offset, node_level): "keys": [], } + fetch_calls = [] + + def fake_fetch(addresses, size): + fetch_calls.append((list(addresses), size)) + return [leaf_raw[a] for a in addresses] + monkeypatch.setattr(tree, "_read_node", fake_read_node) - monkeypatch.setattr(tree, "_estimate_leaf_node_size", lambda _: len(leaf_raw[1000])) + monkeypatch.setattr(tree, "_leaf_node_size", lambda: len(leaf_raw[1000])) + tree._fetch_fn = fake_fetch tree._read_children() - assert len(fs.calls) == 1 - paths, starts, stops = fs.calls[0] - assert paths == ["bucket/file.h5"] * 3 - assert starts == [1000, 1001, 2000] - assert stops == [ - 1000 + len(leaf_raw[1000]), - 1001 + len(leaf_raw[1001]), - 2000 + len(leaf_raw[2000]), - ] - + assert fetch_calls == [([1000, 1001, 2000], len(leaf_raw[1000]))] assert 0 in tree.all_nodes assert [node["addresses"][0] for node in tree.all_nodes[0]] == [7000, 7001, 7002] -def test_raw_chunk_tree_falls_back_when_cat_ranges_unsupported(monkeypatch): +def test_read_children_falls_back_when_fetch_fn_none(monkeypatch): tree = BTreeV1RawDataChunks.__new__(BTreeV1RawDataChunks) - tree.fh = object() + tree._fetch_fn = None called = {"value": False} def fake_super_read_children(self): called["value"] = True - monkeypatch.setattr(AbstractBTree, "_read_children", fake_super_read_children) + monkeypatch.setattr(BTreeV1, "_read_children", fake_super_read_children) tree._read_children() assert called["value"] is True -def test_estimate_leaf_node_size_uses_header_entries_used(): +def test_leaf_node_size_uses_first_leaf_header(): header_addr = 32 buf = bytearray(b"\x00" * header_addr) buf.extend(struct.pack("<4sBBHQQ", b"TREE", 1, 0, 3, 0, 0)) @@ -136,6 +100,52 @@ def test_estimate_leaf_node_size_uses_header_entries_used(): tree = BTreeV1RawDataChunks.__new__(BTreeV1RawDataChunks) tree.fh = fh tree.dims = 2 + tree.all_nodes = {1: [{"addresses": [header_addr]}]} # Header (24) + entries_used (3) * entry_size (8 + 16 + 8) - assert tree._estimate_leaf_node_size(header_addr) == 120 + assert tree._leaf_node_size() == 120 + + +def test_make_btree_fetch_fn_cat_ranges_case(): + dsid = DatasetID.__new__(DatasetID) + dsid.posix = False + dsid._cat_range_allowed = True + dsid._thread_count = 0 + + fs = _DummyFS({10: b"abcd", 20: b"efgh"}) + dsid._DatasetID__fh = _WrappedFH(fs, "bucket/file.h5") + + fetch_fn = dsid._make_btree_fetch_fn() + assert fetch_fn is not None + + out = fetch_fn([10, 20], 4) + assert out == [b"abcd", b"efgh"] + assert fs.calls == [(["bucket/file.h5", "bucket/file.h5"], [10, 20], [14, 24])] + + +def test_make_btree_fetch_fn_pread_case(tmp_path): + dsid = DatasetID.__new__(DatasetID) + dsid.posix = True + dsid._cat_range_allowed = False + dsid._thread_count = 2 + payload = b"abcdefghijklmnopqrstuvwxyz" + fpath = tmp_path / "pread.bin" + fpath.write_bytes(payload) + dsid._filename = str(fpath) + + fetch_fn = dsid._make_btree_fetch_fn() + assert fetch_fn is not None + + out = fetch_fn([0, 4, 8], 3) + assert out == [b"abc", b"efg", b"ijk"] + + +def test_make_btree_fetch_fn_serial_case(): + dsid = DatasetID.__new__(DatasetID) + dsid.posix = True + dsid._cat_range_allowed = False + dsid._thread_count = 0 + + fetch_fn = dsid._make_btree_fetch_fn() + assert fetch_fn is None + From 6327aa74d536ba1fb36f46e36dadd4a09730e721 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Wed, 1 Apr 2026 14:30:06 +0100 Subject: [PATCH 07/24] Implementing #issuecomment-4170115848 --- pyfive/btree.py | 61 ++++++++++++++++++++++--------------------------- pyfive/h5d.py | 45 ++++++++++++++++++++++++++++++++++-- 2 files changed, 70 insertions(+), 36 deletions(-) diff --git a/pyfive/btree.py b/pyfive/btree.py index 2aeadf7b..792fb2cb 100644 --- a/pyfive/btree.py +++ b/pyfive/btree.py @@ -55,16 +55,6 @@ def _read_node(self, offset, node_level): self.last_offset = max(offset, self.last_offset) return node - @staticmethod - def _get_cat_ranges_fs(fh): - """Return (fs, path) when cat_ranges is supported, else (None, None).""" - actual_fh = getattr(fh, "fh", fh) - fs = getattr(actual_fh, "fs", None) - path = getattr(actual_fh, "path", None) - if fs is not None and path is not None and hasattr(fs, "cat_ranges"): - return fs, path - return None, None - def _read_node_header(self, offset): """Return a single node header in the b-tree located at a give offset.""" raise NotImplementedError @@ -136,15 +126,15 @@ class BTreeV1RawDataChunks(BTreeV1): NODE_TYPE = 1 # type: ignore[assignment] - def __init__(self, fh, offset, dims): + def __init__(self, fh, offset, dims, fetch_fn=None): """initalize.""" self.dims = dims + self._fetch_fn = fetch_fn super().__init__(fh, offset) def _read_children(self): - """Read child nodes; fetch leaf-level nodes in bulk when possible.""" - fs, path = self._get_cat_ranges_fs(self.fh) - if fs is None: + """Read children; fetch leaf nodes via fetch_fn when provided.""" + if self._fetch_fn is None: super()._read_children() return @@ -168,13 +158,11 @@ def _read_children(self): if not leaf_addresses: return - leaf_size = self._estimate_leaf_node_size(leaf_addresses[0]) - starts = list(leaf_addresses) - stops = [addr + leaf_size for addr in starts] - raw_leaves = fs.cat_ranges([path] * len(starts), starts, stops) + leaf_size = self._leaf_node_size() + raw_buffers = self._fetch_fn(leaf_addresses, leaf_size) - for addr, raw in zip(starts, raw_leaves): - node = self._parse_leaf_from_buffer(raw, addr) + for addr, raw in zip(leaf_addresses, raw_buffers): + node = self._parse_node_from_buffer(raw, addr, node_level=0) self._add_node(node) def _read_node(self, offset, node_level): @@ -217,32 +205,37 @@ def _read_node(self, offset, node_level): self.last_offset = max(offset, self.last_offset) return node - def _estimate_leaf_node_size(self, first_leaf_addr): - """Estimate leaf-node byte size using the first leaf header.""" - node = self._read_node_header(first_leaf_addr, 0) + def _leaf_node_size(self): + """Compute the byte size of a leaf node by peeking at the first one.""" header_size = struct.calcsize("<" + "".join(self.B_LINK_NODE.values())) - entry_size = 8 + (8 * self.dims) + 8 - return header_size + (node["entries_used"] * entry_size) + entry_size = 8 + 8 * self.dims + 8 + + first_leaf_addr = self.all_nodes[1][0]["addresses"][0] + self.fh.seek(first_leaf_addr) + header_bytes = self.fh.read(header_size) + entries_used = struct.unpack_from(" Date: Wed, 1 Apr 2026 16:46:24 +0100 Subject: [PATCH 08/24] Tiny optimisation in b-tree reading --- pyfive/btree.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/pyfive/btree.py b/pyfive/btree.py index 792fb2cb..fa667fc5 100644 --- a/pyfive/btree.py +++ b/pyfive/btree.py @@ -176,6 +176,8 @@ def _read_node(self, offset, node_level): read_size = n_entries * entry_size node_data = self.fh.read(read_size) mem_view = memoryview(node_data) + offset_fmt = f"<{self.dims}Q" + offset_size = self.dims * 8 pos = 0 for _ in range(n_entries): @@ -183,9 +185,8 @@ def _read_node(self, offset, node_level): chunk_size, filter_mask = struct.unpack_from(" Date: Wed, 1 Apr 2026 17:03:39 +0100 Subject: [PATCH 09/24] Make dataset an ABC so other packages can register versions without subclassing --- pyfive/high_level.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pyfive/high_level.py b/pyfive/high_level.py index fe650b55..f443f67b 100644 --- a/pyfive/high_level.py +++ b/pyfive/high_level.py @@ -4,6 +4,7 @@ from collections import deque from collections.abc import Callable from collections.abc import Mapping, Sequence +from abc import ABC import os import posixpath import warnings @@ -371,7 +372,7 @@ def __exit__(self, exc_type, value, traceback): self.close() -class Dataset(object): +class Dataset(ABC): """ A HDF5 Dataset containing an n-dimensional array and meta-data attributes. From 6b5ba3e32eb7e5f783b1e8453c04195a79773138 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Wed, 1 Apr 2026 17:18:54 +0100 Subject: [PATCH 10/24] Making parallelism the default, the benefits are too incredible --- pyfive/h5d.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyfive/h5d.py b/pyfive/h5d.py index 0f5be56a..433cd8dd 100644 --- a/pyfive/h5d.py +++ b/pyfive/h5d.py @@ -46,7 +46,7 @@ class ChunkRead: # Shared helpers # # ------------------------------------------------------------------ # - def set_parallelism(self, thread_count=0, cat_range_allowed=False): + def set_parallelism(self, thread_count=5, cat_range_allowed=True): """ Configure experimental chunk-read parallelism. From 56f3fb2110a5cea88ced61a932b30b1b6617c4d3 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Mon, 6 Apr 2026 09:46:13 +0100 Subject: [PATCH 11/24] testing for parallel --- pyfive/btree.py | 21 ++- tests/test_btree_parallel.py | 158 ++++++++++++++++++++- tests/test_s3_caching.py | 266 +++++++++++++++++++++++++++++++++++ 3 files changed, 439 insertions(+), 6 deletions(-) create mode 100644 tests/test_s3_caching.py diff --git a/pyfive/btree.py b/pyfive/btree.py index fa667fc5..b3e1adb3 100644 --- a/pyfive/btree.py +++ b/pyfive/btree.py @@ -158,12 +158,23 @@ def _read_children(self): if not leaf_addresses: return - leaf_size = self._leaf_node_size() - raw_buffers = self._fetch_fn(leaf_addresses, leaf_size) + # Leaf nodes can differ in entries_used, so fetch just headers first, + # then group addresses by exact node size. + header_size = struct.calcsize("<" + "".join(self.B_LINK_NODE.values())) + header_buffers = self._fetch_fn(leaf_addresses, header_size) - for addr, raw in zip(leaf_addresses, raw_buffers): - node = self._parse_node_from_buffer(raw, addr, node_level=0) - self._add_node(node) + size_to_addresses = OrderedDict() + entry_size = 8 + self.dims * 8 + 8 + for addr, header in zip(leaf_addresses, header_buffers): + entries_used = struct.unpack_from(" 0 else b'' + + def seek(self, offset): + return offset + + def close(self): + pass + + # Mock fsspec to use our MockS3File + with patch('fsspec.open', return_value=MagicMock(__enter__=lambda self: MockS3File())): + if access_pattern == "sequential": + # Each access creates a new file + for i in range(3): + try: + f = MockS3File() + f.read(100) + except Exception: + pass + else: # repeated + # Same file accessed multiple times + f = MockS3File() + for i in range(3): + try: + f.read(100) + except Exception: + pass + + # Verify that in "repeated" pattern, the same handle is used + if access_pattern == "repeated": + assert len(set(r['handle_id'] for r in read_operations)) == 1, \ + "Expected same handle for repeated access" + + +class TestDuplicateAttributeReads: + """ + Test to understand if attributes are read twice per variable. + + Based on user's observation that every variable is read twice: + - First via collect_dimensions_from_root() + - Second via dump_header() + + This investigates whether those are separate DataObjects instances + or cached reads of the same data. + """ + + def test_duplicate_attribute_read_detection(self, caplog): + """ + Test framework to detect duplicate attribute reads. + + Logs should show if same file handle is reading same offset twice. + """ + # This is a framework test - in practice, run with: + # python -m pytest tests/test_s3_caching.py::TestDuplicateAttributeReads::test_duplicate_attribute_read_detection -v -s --log-cli-level=DEBUG + + # Expected behavior to verify: + # 1. If fh_id is SAME both times and offset is SAME but timing is DIFFERENT + # → fsspec cache NOT working properly (both are misses) + # 2. If fh_id is SAME, offset is SAME, timing is FAST second time + # → fsspec cache IS working (first miss, second hit) + # 3. If fh_id is DIFFERENT each time + # → file handles being recreated unnecessarily + + assert True, "See log output for file handle reuse patterns" + + +class TestReadAheadCacheStatistics: + """ + Verify fsspec readahead cache statistics. + + From the user's log: + "readahead: 11300 hits, 32 misses" + + This means: + - 11,300 cached reads served + - 32 cache misses (actual network reads) + - Very high cache hit rate: 99.7% + """ + + def test_readahead_cache_effectiveness(self): + """ + Document the readahead cache behavior observed. + + The fsspec readahead cache shows: + - High hit rate (99.7%) when accessing same S3 file multiple times + - Small byte ranges (8-40 bytes) for attribute data + - Cache spans across multiple variable attribute reads + + Implication: + - If file handle is REUSED: cache working well, slowness is network latency + - If file handle is RECREATED: cache would start fresh (different story) + """ + cache_hits = 11300 + cache_misses = 32 + total_requests = cache_hits + cache_misses + hit_rate = cache_hits / total_requests + + # Document observed behavior + assert hit_rate > 0.99, f"Cache hit rate {hit_rate:.1%} is excellent" + + +class TestFileHandleIdTracking: + """ + Tests to verify the diagnostic logging in DataObjects.__init__ + and DataObjects.get_attributes() is working. + """ + + def test_dataobjects_init_logs_handle_id(self, caplog): + """Test that DataObjects.__init__ logs file handle ID.""" + test_data = io.BytesIO(b'\x89HDF\r\n\x1a\n' + b'\x00' * 100) + test_data.seek(0) + + with caplog.at_level(logging.DEBUG, logger='pyfive'): + try: + do = DataObjects(test_data, 0) + except Exception: + pass + + # Look for the diagnostic log with fh_id, type, s3, offset + # Format: "[pyfive] DataObjects init: fh_id= type= s3= offset=" + logs = [r.getMessage() for r in caplog.records if 'DataObjects init' in r.getMessage()] + # Framework in place to capture this + + +if __name__ == '__main__': + pytest.main([__file__, '-v', '-s']) From a3cc1219cd0a2cec7659de9612a9dbdba5e03377 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Mon, 6 Apr 2026 10:22:33 +0100 Subject: [PATCH 12/24] UML for pyfive in this branch --- doc/pyfive_class_diagram.pu | 158 ++++++++++++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) create mode 100644 doc/pyfive_class_diagram.pu diff --git a/doc/pyfive_class_diagram.pu b/doc/pyfive_class_diagram.pu new file mode 100644 index 00000000..e99b118a --- /dev/null +++ b/doc/pyfive_class_diagram.pu @@ -0,0 +1,158 @@ +@startuml +skinparam classAttributeIconSize 0 +'skinparam linetype ortho +'Use portrait layout +top to bottom direction +skinparam ClassBackgroundColor palegoldenrod +skinparam ClassBorderColor Black + + + +skinparam stereotypeCBackgroundColor white + + + +!define BTREE_COLOR white +!define BTREE_STEREO <<(B,BTREE_COLOR)>> +!define HEAP_COLOR white +!define HEAP_STEREO <<(H,HEAP_COLOR)>> + + +' ------------------------------------------------------------------ +' Packages +' ------------------------------------------------------------------ + +package "pyfive.high_level" { + class File { + + filename : str + + mode : str + + __init__(filename, mode, metadata_buffer_size) + + close() + + __enter__() + + __exit__() + + consolidated_metadata : bool + } + + class Group { + + name : str + + attrs : dict + + __init__(name, dataobjects, parent) + + __getitem__(name) + + visit(func) + + visititems(func, noindex) + + get_lazy_view(name) + } + + class Dataset { + + name : str + + shape : tuple + + dtype : dtype + + size : int + + chunks : tuple + + __init__(name, datasetid, parent) + + __getitem__(args) + } +} + +package "pyfive.h5d" { + class ChunkRead { + + set_parallelism(thread_count, cat_range_allowed) + + _select_chunks(indexer, out, dtype) + } + + class DatasetID { + + rank : int + + shape : tuple + + dtype : dtype + + chunks : tuple + + layout_class : int + + __init__(dataobjects) + + get_data(args, fillvalue) + + get_chunk_info(index) + + get_num_chunks() + + _build_index() + + _make_btree_fetch_fn() + } + + ChunkRead <|- DatasetID +} + +package "pyfive.dataobjects" { + class DataObjects { + + offset : int + + is_group : bool + + is_dataset : bool + + get_attributes() + + get_links() + } +} + +package "pyfive.misc_low_level" { + class SuperBlock { + + version : int + + offset_to_dataobjects : int + } + + class Heap HEAP_STEREO + class GlobalHeap HEAP_STEREO + class FractalHeap HEAP_STEREO + class SymbolTable +} + +package "pyfive.btree" { + class AbstractBTree BTREE_STEREO + class BTreeV1 BTREE_STEREO + class BTreeV1\nGroups BTREE_STEREO + class BTreeV1\nRawDataChunks BTREE_STEREO + class BTreeV2 BTREE_STEREO + class BTreeV2\nGroupNames BTREE_STEREO + class BTreeV2\nGroupOrders BTREE_STEREO + class BTreeV2\nAttrCreationOrder BTREE_STEREO + class BTreeV2\nAttrNames BTREE_STEREO + + BTreeV1 --|> AbstractBTree + BTreeV2 --|> AbstractBTree + BTreeV1\nGroups --|> BTreeV1 + BTreeV1\nRawDataChunks --|> BTreeV1 + BTreeV2\nGroupNames --|> BTreeV2 + BTreeV2\nGroupOrders --|> BTreeV2 + BTreeV2\nAttrCreationOrder --|> BTreeV2 + BTreeV2\nAttrNames --|> BTreeV2 +} + +package "pyfive.utilities" { + class MetadataBufferingWrapper +} + +' ------------------------------------------------------------------ +' Relationships +' ------------------------------------------------------------------ + +' High-Level Hierarchy +File --|> Group +Group o-- Dataset +Group o-- Group + +' Composition +File *-- SuperBlock : parses > +SuperBlock --> DataObjects : points to (root) +Group *-- DataObjects : uses > +Dataset *-- DatasetID : wraps > +DatasetID --> DataObjects : uses metadata > + +' Lower Level internal mechanics +DataObjects ..> Heap : reads names +DataObjects ..> GlobalHeap : reads strings +DataObjects ..> FractalHeap : reads messages +DataObjects ..> SymbolTable : reads entries + +' B-Tree usage +DataObjects ..> BTreeV1\nGroups : iterates children +DataObjects ..> BTreeV2\nGroupNames : iterates children +DataObjects ..> BTreeV2\nGroupOrders : iterates children +DataObjects ..> BTreeV2\nAttrCreationOrder : iterates attrs +DataObjects ..> BTreeV2\nAttrNames : iterates attrs +DatasetID ..> BTreeV1\nRawDataChunks : finds data chunks +File ..> MetadataBufferingWrapper : wraps remote handles + +@enduml From 6a4a0ef7ae37d30375c9e71c3a70468afad8357e Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Tue, 7 Apr 2026 13:41:40 +0100 Subject: [PATCH 13/24] revised class diagram --- doc/pyfive_class_diagram.pu | 46 +++++++++++++++++++++++++++++-------- 1 file changed, 37 insertions(+), 9 deletions(-) diff --git a/doc/pyfive_class_diagram.pu b/doc/pyfive_class_diagram.pu index e99b118a..6919a6e5 100644 --- a/doc/pyfive_class_diagram.pu +++ b/doc/pyfive_class_diagram.pu @@ -22,7 +22,9 @@ skinparam stereotypeCBackgroundColor white ' Packages ' ------------------------------------------------------------------ -package "pyfive.high_level" { +package pyfive { + +package "high_level" { class File { + filename : str + mode : str @@ -54,7 +56,7 @@ package "pyfive.high_level" { } } -package "pyfive.h5d" { +package "h5d" { class ChunkRead { + set_parallelism(thread_count, cat_range_allowed) + _select_chunks(indexer, out, dtype) @@ -77,7 +79,7 @@ package "pyfive.h5d" { ChunkRead <|- DatasetID } -package "pyfive.dataobjects" { +package "dataobjects" { class DataObjects { + offset : int + is_group : bool @@ -87,7 +89,7 @@ package "pyfive.dataobjects" { } } -package "pyfive.misc_low_level" { +package "misc_low_level" { class SuperBlock { + version : int + offset_to_dataobjects : int @@ -97,13 +99,22 @@ package "pyfive.misc_low_level" { class GlobalHeap HEAP_STEREO class FractalHeap HEAP_STEREO class SymbolTable + + SymbolTable -[hidden]- Heap + SymbolTable -[hidden]- GlobalHeap + SymbolTable -[hidden]- FractalHeap } -package "pyfive.btree" { +package "btree" { class AbstractBTree BTREE_STEREO class BTreeV1 BTREE_STEREO class BTreeV1\nGroups BTREE_STEREO - class BTreeV1\nRawDataChunks BTREE_STEREO + class BTreeV1\nRawDataChunks BTREE_STEREO { + + fh : int + + offset : int + + dims : int + fetch_fn() + } class BTreeV2 BTREE_STEREO class BTreeV2\nGroupNames BTREE_STEREO class BTreeV2\nGroupOrders BTREE_STEREO @@ -116,11 +127,13 @@ package "pyfive.btree" { BTreeV1\nRawDataChunks --|> BTreeV1 BTreeV2\nGroupNames --|> BTreeV2 BTreeV2\nGroupOrders --|> BTreeV2 - BTreeV2\nAttrCreationOrder --|> BTreeV2 + BTreeV2\nAttrCreationOrder -|> BTreeV2 BTreeV2\nAttrNames --|> BTreeV2 + + } -package "pyfive.utilities" { +package "utilities" { class MetadataBufferingWrapper } @@ -152,7 +165,22 @@ DataObjects ..> BTreeV2\nGroupNames : iterates children DataObjects ..> BTreeV2\nGroupOrders : iterates children DataObjects ..> BTreeV2\nAttrCreationOrder : iterates attrs DataObjects ..> BTreeV2\nAttrNames : iterates attrs -DatasetID ..> BTreeV1\nRawDataChunks : finds data chunks +DatasetID ..> BTreeV1\nRawDataChunks : finds\ndata\nchunks File ..> MetadataBufferingWrapper : wraps remote handles +note as N1 +b-tree +- built by _build_index +- parallelism passed + via the fetch_fn to + BTreeV1RawDataChunks +end note + +DatasetID .. N1 +N1 .. BTreeV1\nRawDataChunks + +} + + + @enduml From 548e1303d6a500d8b4d216f6c863bd01df6a2756 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Tue, 7 Apr 2026 13:59:15 +0100 Subject: [PATCH 14/24] b-tree parallelism off by default --- pyfive/h5d.py | 16 +++++++++++++--- tests/test_btree_parallel.py | 21 +++++++++------------ 2 files changed, 22 insertions(+), 15 deletions(-) diff --git a/pyfive/h5d.py b/pyfive/h5d.py index 433cd8dd..6554f458 100644 --- a/pyfive/h5d.py +++ b/pyfive/h5d.py @@ -46,16 +46,20 @@ class ChunkRead: # Shared helpers # # ------------------------------------------------------------------ # - def set_parallelism(self, thread_count=5, cat_range_allowed=True): + def set_parallelism(self, thread_count=5, cat_range_allowed=True, _btree_parallel=False): """ Configure experimental chunk-read parallelism. ``thread_count`` controls POSIX threaded reads via ``os.pread``: - ``0`` disables threaded reads - ``>0`` enables threaded reads with that many workers + - Default 5 ``cat_range_allowed`` enables fsspec bulk reads via ``cat_ranges`` - for compatible non-posix file handles. + for compatible non-posix file handles. Default True + + ``parallel_btree`` enables parallel reads for b-tree nodes when building + the chunk index. Default False. This is a ``pyfive`` API extension, and is opt-in by default as it may not be suitable for all use cases. It is recommended to enable it when working with remote files, but it may not be suitable for local files. @@ -69,7 +73,9 @@ def set_parallelism(self, thread_count=5, cat_range_allowed=True): self._thread_count = thread_count self._cat_range_allowed = bool(cat_range_allowed) - logger.info('Parallelism set with thread_count=%d and cat_range_allowed=%s', self._thread_count, self._cat_range_allowed) + self._btree_parallel = bool(_btree_parallel) + logger.info('Parallelism: thread_count=%d, cat_range_allowed=%s, btree_parallel=%s', + self._thread_count, self._cat_range_allowed, self._btree_parallel) def _get_required_chunks(self, indexer): @@ -613,6 +619,10 @@ def _build_index(self): def _make_btree_fetch_fn(self): """Return fetch_fn(addresses, size) for b-tree leaf reads, or None.""" + + if not self._btree_parallel: + return None + actual_fh = None if not self.posix: fh = self._fh diff --git a/tests/test_btree_parallel.py b/tests/test_btree_parallel.py index 5a5e77f5..79df0110 100644 --- a/tests/test_btree_parallel.py +++ b/tests/test_btree_parallel.py @@ -166,8 +166,7 @@ def test_leaf_node_size_uses_first_leaf_header(): def test_make_btree_fetch_fn_cat_ranges_case(): dsid = DatasetID.__new__(DatasetID) dsid.posix = False - dsid._cat_range_allowed = True - dsid._thread_count = 0 + dsid.set_parallelism(thread_count=0, cat_range_allowed=True, _btree_parallel=True) fs = _DummyFS({10: b"abcd", 20: b"efgh"}) dsid._DatasetID__fh = _WrappedFH(fs, "bucket/file.h5") @@ -183,8 +182,7 @@ def test_make_btree_fetch_fn_cat_ranges_case(): def test_make_btree_fetch_fn_pread_case(tmp_path): dsid = DatasetID.__new__(DatasetID) dsid.posix = True - dsid._cat_range_allowed = False - dsid._thread_count = 2 + dsid.set_parallelism(thread_count=2, cat_range_allowed=False, _btree_parallel=True) payload = b"abcdefghijklmnopqrstuvwxyz" fpath = tmp_path / "pread.bin" fpath.write_bytes(payload) @@ -200,8 +198,7 @@ def test_make_btree_fetch_fn_pread_case(tmp_path): def test_make_btree_fetch_fn_serial_case(): dsid = DatasetID.__new__(DatasetID) dsid.posix = True - dsid._cat_range_allowed = False - dsid._thread_count = 0 + dsid.set_parallelism(thread_count=0, cat_range_allowed=False, _btree_parallel=False) fetch_fn = dsid._make_btree_fetch_fn() assert fetch_fn is None @@ -210,7 +207,7 @@ def test_make_btree_fetch_fn_serial_case(): def test_parallel_pread_matches_serial_results_for_chunked_dataset(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False) + serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, _btree_parallel=False) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -218,7 +215,7 @@ def test_parallel_pread_matches_serial_results_for_chunked_dataset(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: parallel = hfile["dataset1"] - parallel.id.set_parallelism(thread_count=2, cat_range_allowed=False) + parallel.id.set_parallelism(thread_count=2, cat_range_allowed=False, _btree_parallel=True) parallel_data = parallel[:] parallel_chunk_info = [ parallel.id.get_chunk_info(i) for i in range(parallel.id.get_num_chunks()) @@ -231,7 +228,7 @@ def test_parallel_pread_matches_serial_results_for_chunked_dataset(): def test_parallel_fsspec_cat_ranges_matches_serial_results(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False) + serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, _btree_parallel=False) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -254,7 +251,7 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): with memfs.open(mem_path, "rb") as fh: with pyfive.File(fh) as hfile: ds = hfile["dataset1"] - ds.id.set_parallelism(thread_count=0, cat_range_allowed=True) + ds.id.set_parallelism(thread_count=0, cat_range_allowed=True, _btree_parallel=True) fsspec_data = ds[:] fsspec_chunk_info = [ ds.id.get_chunk_info(i) for i in range(ds.id.get_num_chunks()) @@ -268,7 +265,7 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): def test_parallel_s3fs_cat_ranges_matches_serial_results(s3fs_s3): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False) + serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, _btree_parallel=False) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -295,7 +292,7 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): with s3fs_s3.open(s3_key, "rb") as fh: with pyfive.File(fh) as hfile: ds = hfile["dataset1"] - ds.id.set_parallelism(thread_count=0, cat_range_allowed=True) + ds.id.set_parallelism(thread_count=0, cat_range_allowed=True, _btree_parallel=True) s3_data = ds[:] s3_chunk_info = [ ds.id.get_chunk_info(i) for i in range(ds.id.get_num_chunks()) From c518cb83ec43649b804ca333fa7603842596b793 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Wed, 8 Apr 2026 08:30:24 +0100 Subject: [PATCH 15/24] Fix ruff problem --- tests/test_s3_caching.py | 79 ++++++++++++++++++++-------------------- 1 file changed, 40 insertions(+), 39 deletions(-) diff --git a/tests/test_s3_caching.py b/tests/test_s3_caching.py index 56ba57e0..324adff4 100644 --- a/tests/test_s3_caching.py +++ b/tests/test_s3_caching.py @@ -9,6 +9,7 @@ import logging from unittest.mock import Mock, patch, MagicMock from pathlib import Path +from pyfive.dataobjects import DataObjects import pytest @@ -20,32 +21,32 @@ class TestFileHandleReuse: """Test whether file handles are reused across DataObjects instances.""" - + def test_dataobjects_tracks_file_handle_id(self, tmp_path): """Test that DataObjects tracks file handle identity.""" from pyfive.dataobjects import DataObjects - + # Create a simple HDF5 file test_file = tmp_path / "test.hdf5" - + # Create a minimal HDF5 file by writing bytes with open(test_file, 'wb') as f: # HDF5 signature f.write(b'\x89HDF\r\n\x1a\n') # Minimal superblock (rest can be zeros for this test) f.write(b'\x00' * 100) - + # Open file and create DataObjects instances with open(test_file, 'rb') as fh: fh_id_first = id(fh) - + # Create first DataObjects instance try: do1 = DataObjects(fh, 0) except Exception: # May fail to parse, but that's okay - we're testing handle tracking pass - + # The file handle should be the same object assert id(fh) == fh_id_first @@ -53,20 +54,20 @@ def test_dataobjects_tracks_file_handle_id(self, tmp_path): def test_file_handle_identity_in_logging(self, caplog, tmp_path): """Test that file handle ID is logged consistently.""" from pyfive.dataobjects import DataObjects - + test_file = tmp_path / "test.hdf5" - + with open(test_file, 'wb') as f: f.write(b'\x89HDF\r\n\x1a\n') f.write(b'\x00' * 100) - + with caplog.at_level(logging.DEBUG, logger='pyfive'): with open(test_file, 'rb') as fh: try: do1 = DataObjects(fh, 0) except Exception: pass - + # Check if any pyfive diagnostic logs were captured pyfive_logs = [r for r in caplog.records if 'pyfive' in r.getMessage()] # Just verify we can create DataObjects without error @@ -74,19 +75,19 @@ def test_file_handle_identity_in_logging(self, caplog, tmp_path): class TestAttributeReadCaching: """Test whether attributes are being re-read or cached.""" - + def test_get_attributes_logs_file_handle_info(self, caplog, tmp_path): """Verify that get_attributes() logs file handle and type information.""" from pyfive.dataobjects import DataObjects - + # This test verifies the logging infrastructure is in place # to track whether attributes are being cached - + test_file = tmp_path / "test.hdf5" with open(test_file, 'wb') as f: f.write(b'\x89HDF\r\n\x1a\n') f.write(b'\x00' * 100) - + with caplog.at_level(logging.INFO, logger='pyfive'): with open(test_file, 'rb') as fh: try: @@ -96,7 +97,7 @@ def test_get_attributes_logs_file_handle_info(self, caplog, tmp_path): do.get_attributes() except Exception: pass - + # Check for expected log message pattern # Format: "[pyfive] Obtained N attributes from offset Y (fh_id=X type=Y) in Z.XXXXs" logs_with_obtained = [ @@ -108,7 +109,7 @@ def test_get_attributes_logs_file_handle_info(self, caplog, tmp_path): class TestS3FileHandleLifecycle: """Test S3 file handle lifecycle and reuse patterns.""" - + @pytest.mark.parametrize("access_pattern", [ "sequential", # Access each variable once "repeated", # Access same variables multiple times @@ -116,37 +117,37 @@ class TestS3FileHandleLifecycle: def test_s3_file_handle_reuse(self, access_pattern): """ Test whether S3 file handles are reused across variable accesses. - + This test uses mocking to simulate S3 access and verify file handle identity is maintained or new handles are created as expected. """ # Track file handle IDs and read operations file_handle_ids = [] read_operations = [] - + class MockS3File: """Mock S3 file that tracks handle creation and reads.""" _instance_counter = 0 - + def __init__(self, *args, **kwargs): MockS3File._instance_counter += 1 self.id = MockS3File._instance_counter self.fs = MagicMock() # Simulate S3FileSystem file_handle_ids.append(self.id) - + def read(self, size=-1): read_operations.append({ 'handle_id': self.id, 'size': size }) return b'\x00' * size if size > 0 else b'' - + def seek(self, offset): return offset - + def close(self): pass - + # Mock fsspec to use our MockS3File with patch('fsspec.open', return_value=MagicMock(__enter__=lambda self: MockS3File())): if access_pattern == "sequential": @@ -165,7 +166,7 @@ def close(self): f.read(100) except Exception: pass - + # Verify that in "repeated" pattern, the same handle is used if access_pattern == "repeated": assert len(set(r['handle_id'] for r in read_operations)) == 1, \ @@ -175,24 +176,24 @@ def close(self): class TestDuplicateAttributeReads: """ Test to understand if attributes are read twice per variable. - + Based on user's observation that every variable is read twice: - First via collect_dimensions_from_root() - Second via dump_header() - + This investigates whether those are separate DataObjects instances or cached reads of the same data. """ - + def test_duplicate_attribute_read_detection(self, caplog): """ Test framework to detect duplicate attribute reads. - + Logs should show if same file handle is reading same offset twice. """ # This is a framework test - in practice, run with: # python -m pytest tests/test_s3_caching.py::TestDuplicateAttributeReads::test_duplicate_attribute_read_detection -v -s --log-cli-level=DEBUG - + # Expected behavior to verify: # 1. If fh_id is SAME both times and offset is SAME but timing is DIFFERENT # → fsspec cache NOT working properly (both are misses) @@ -200,32 +201,32 @@ def test_duplicate_attribute_read_detection(self, caplog): # → fsspec cache IS working (first miss, second hit) # 3. If fh_id is DIFFERENT each time # → file handles being recreated unnecessarily - + assert True, "See log output for file handle reuse patterns" class TestReadAheadCacheStatistics: """ Verify fsspec readahead cache statistics. - + From the user's log: "readahead: 11300 hits, 32 misses" - + This means: - 11,300 cached reads served - 32 cache misses (actual network reads) - Very high cache hit rate: 99.7% """ - + def test_readahead_cache_effectiveness(self): """ Document the readahead cache behavior observed. - + The fsspec readahead cache shows: - High hit rate (99.7%) when accessing same S3 file multiple times - Small byte ranges (8-40 bytes) for attribute data - Cache spans across multiple variable attribute reads - + Implication: - If file handle is REUSED: cache working well, slowness is network latency - If file handle is RECREATED: cache would start fresh (different story) @@ -234,7 +235,7 @@ def test_readahead_cache_effectiveness(self): cache_misses = 32 total_requests = cache_hits + cache_misses hit_rate = cache_hits / total_requests - + # Document observed behavior assert hit_rate > 0.99, f"Cache hit rate {hit_rate:.1%} is excellent" @@ -244,18 +245,18 @@ class TestFileHandleIdTracking: Tests to verify the diagnostic logging in DataObjects.__init__ and DataObjects.get_attributes() is working. """ - + def test_dataobjects_init_logs_handle_id(self, caplog): """Test that DataObjects.__init__ logs file handle ID.""" test_data = io.BytesIO(b'\x89HDF\r\n\x1a\n' + b'\x00' * 100) test_data.seek(0) - + with caplog.at_level(logging.DEBUG, logger='pyfive'): try: do = DataObjects(test_data, 0) except Exception: pass - + # Look for the diagnostic log with fh_id, type, s3, offset # Format: "[pyfive] DataObjects init: fh_id= type= s3= offset=" logs = [r.getMessage() for r in caplog.records if 'DataObjects init' in r.getMessage()] From 4fe2831d3e53c365f0cd2ea390efce4c5d998901 Mon Sep 17 00:00:00 2001 From: Valeriu Predoi Date: Thu, 9 Apr 2026 16:45:59 +0100 Subject: [PATCH 16/24] run pre-commit --- pyfive/h5d.py | 30 +++++++++++++++++++++--------- pyfive/utilities.py | 6 +++--- 2 files changed, 24 insertions(+), 12 deletions(-) diff --git a/pyfive/h5d.py b/pyfive/h5d.py index 6554f458..7dd9d786 100644 --- a/pyfive/h5d.py +++ b/pyfive/h5d.py @@ -26,7 +26,6 @@ ChunkIndex = namedtuple("ChunkIndex", "chunk_address chunk_dims") - class ChunkRead: """ Mixin providing parallel and bulk chunk-reading strategies. @@ -46,7 +45,9 @@ class ChunkRead: # Shared helpers # # ------------------------------------------------------------------ # - def set_parallelism(self, thread_count=5, cat_range_allowed=True, _btree_parallel=False): + def set_parallelism( + self, thread_count=5, cat_range_allowed=True, _btree_parallel=False + ): """ Configure experimental chunk-read parallelism. @@ -58,7 +59,7 @@ def set_parallelism(self, thread_count=5, cat_range_allowed=True, _btree_paralle ``cat_range_allowed`` enables fsspec bulk reads via ``cat_ranges`` for compatible non-posix file handles. Default True - ``parallel_btree`` enables parallel reads for b-tree nodes when building + ``parallel_btree`` enables parallel reads for b-tree nodes when building the chunk index. Default False. This is a ``pyfive`` API extension, and is opt-in by default as it may not be suitable for all use cases. @@ -74,9 +75,12 @@ def set_parallelism(self, thread_count=5, cat_range_allowed=True, _btree_paralle self._thread_count = thread_count self._cat_range_allowed = bool(cat_range_allowed) self._btree_parallel = bool(_btree_parallel) - logger.info('Parallelism: thread_count=%d, cat_range_allowed=%s, btree_parallel=%s', - self._thread_count, self._cat_range_allowed, self._btree_parallel) - + logger.info( + "Parallelism: thread_count=%d, cat_range_allowed=%s, btree_parallel=%s", + self._thread_count, + self._cat_range_allowed, + self._btree_parallel, + ) def _get_required_chunks(self, indexer): """ @@ -120,7 +124,9 @@ def _select_chunks(self, indexer, out, dtype): fh = self._fh actual_fh = getattr(fh, "fh", fh) # support wrapped file-like objects if hasattr(actual_fh, "fs") and hasattr(actual_fh.fs, "cat_ranges"): - logger.info(f"[pyfive] chunk read strategy: fsspec_cat_ranges ({len(chunks)} chunks)") + logger.info( + f"[pyfive] chunk read strategy: fsspec_cat_ranges ({len(chunks)} chunks)" + ) self._read_bulk_fsspec(fh, chunks, out, dtype) return @@ -181,7 +187,9 @@ def _read_one(item): finally: fh.close() - logger.info('pyfive thread pool read completed using %d threads', self._thread_count) + logger.info( + "pyfive thread pool read completed using %d threads", self._thread_count + ) for chunk_sel, out_sel, filter_mask, chunk_buffer in results: out[out_sel] = self._decode_chunk(chunk_buffer, filter_mask, dtype)[ @@ -650,7 +658,11 @@ def fetch_pread(addresses, size): fd = fh_local.fileno() try: with ThreadPoolExecutor(max_workers=thread_count) as executor: - return list(executor.map(lambda addr: os.pread(fd, size, addr), addresses)) + return list( + executor.map( + lambda addr: os.pread(fd, size, addr), addresses + ) + ) finally: fh_local.close() diff --git a/pyfive/utilities.py b/pyfive/utilities.py index 7471be1f..3089f92d 100644 --- a/pyfive/utilities.py +++ b/pyfive/utilities.py @@ -79,13 +79,13 @@ def closed(self) -> bool: @property def fs(self): - return getattr(self.fh, "fs", None) + return getattr(self.fh, "fs", None) @property def path(self): - return getattr(self.fh, "path", None) + return getattr(self.fh, "path", None) - def __getattr__(self,name:str): + def __getattr__(self, name: str): return getattr(self.fh, name) def _ensure_buffer(self): From 530c66ae4d7ea101af700226fe34fcc916ab96ed Mon Sep 17 00:00:00 2001 From: Valeriu Predoi Date: Thu, 9 Apr 2026 16:46:25 +0100 Subject: [PATCH 17/24] run pre-commit on two test files --- tests/test_btree_parallel.py | 25 +++++++---- tests/test_s3_caching.py | 80 ++++++++++++++++++++---------------- 2 files changed, 62 insertions(+), 43 deletions(-) diff --git a/tests/test_btree_parallel.py b/tests/test_btree_parallel.py index 79df0110..737e70f0 100644 --- a/tests/test_btree_parallel.py +++ b/tests/test_btree_parallel.py @@ -207,7 +207,9 @@ def test_make_btree_fetch_fn_serial_case(): def test_parallel_pread_matches_serial_results_for_chunked_dataset(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, _btree_parallel=False) + serial.id.set_parallelism( + thread_count=0, cat_range_allowed=False, _btree_parallel=False + ) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -215,7 +217,9 @@ def test_parallel_pread_matches_serial_results_for_chunked_dataset(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: parallel = hfile["dataset1"] - parallel.id.set_parallelism(thread_count=2, cat_range_allowed=False, _btree_parallel=True) + parallel.id.set_parallelism( + thread_count=2, cat_range_allowed=False, _btree_parallel=True + ) parallel_data = parallel[:] parallel_chunk_info = [ parallel.id.get_chunk_info(i) for i in range(parallel.id.get_num_chunks()) @@ -228,7 +232,9 @@ def test_parallel_pread_matches_serial_results_for_chunked_dataset(): def test_parallel_fsspec_cat_ranges_matches_serial_results(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, _btree_parallel=False) + serial.id.set_parallelism( + thread_count=0, cat_range_allowed=False, _btree_parallel=False + ) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -251,7 +257,9 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): with memfs.open(mem_path, "rb") as fh: with pyfive.File(fh) as hfile: ds = hfile["dataset1"] - ds.id.set_parallelism(thread_count=0, cat_range_allowed=True, _btree_parallel=True) + ds.id.set_parallelism( + thread_count=0, cat_range_allowed=True, _btree_parallel=True + ) fsspec_data = ds[:] fsspec_chunk_info = [ ds.id.get_chunk_info(i) for i in range(ds.id.get_num_chunks()) @@ -265,7 +273,9 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): def test_parallel_s3fs_cat_ranges_matches_serial_results(s3fs_s3): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, _btree_parallel=False) + serial.id.set_parallelism( + thread_count=0, cat_range_allowed=False, _btree_parallel=False + ) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -292,7 +302,9 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): with s3fs_s3.open(s3_key, "rb") as fh: with pyfive.File(fh) as hfile: ds = hfile["dataset1"] - ds.id.set_parallelism(thread_count=0, cat_range_allowed=True, _btree_parallel=True) + ds.id.set_parallelism( + thread_count=0, cat_range_allowed=True, _btree_parallel=True + ) s3_data = ds[:] s3_chunk_info = [ ds.id.get_chunk_info(i) for i in range(ds.id.get_num_chunks()) @@ -301,4 +313,3 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): assert calls, "Expected s3fs cat_ranges to be used for leaf-node reads" assert_array_equal(s3_data, serial_data) assert s3_chunk_info == serial_chunk_info - diff --git a/tests/test_s3_caching.py b/tests/test_s3_caching.py index 324adff4..30fdf553 100644 --- a/tests/test_s3_caching.py +++ b/tests/test_s3_caching.py @@ -16,7 +16,7 @@ # Configure logging to capture pyfive diagnostic messages logging.basicConfig(level=logging.DEBUG) -logger = logging.getLogger('pyfive') +logger = logging.getLogger("pyfive") class TestFileHandleReuse: @@ -30,14 +30,14 @@ def test_dataobjects_tracks_file_handle_id(self, tmp_path): test_file = tmp_path / "test.hdf5" # Create a minimal HDF5 file by writing bytes - with open(test_file, 'wb') as f: + with open(test_file, "wb") as f: # HDF5 signature - f.write(b'\x89HDF\r\n\x1a\n') + f.write(b"\x89HDF\r\n\x1a\n") # Minimal superblock (rest can be zeros for this test) - f.write(b'\x00' * 100) + f.write(b"\x00" * 100) # Open file and create DataObjects instances - with open(test_file, 'rb') as fh: + with open(test_file, "rb") as fh: fh_id_first = id(fh) # Create first DataObjects instance @@ -50,26 +50,25 @@ def test_dataobjects_tracks_file_handle_id(self, tmp_path): # The file handle should be the same object assert id(fh) == fh_id_first - def test_file_handle_identity_in_logging(self, caplog, tmp_path): """Test that file handle ID is logged consistently.""" from pyfive.dataobjects import DataObjects test_file = tmp_path / "test.hdf5" - with open(test_file, 'wb') as f: - f.write(b'\x89HDF\r\n\x1a\n') - f.write(b'\x00' * 100) + with open(test_file, "wb") as f: + f.write(b"\x89HDF\r\n\x1a\n") + f.write(b"\x00" * 100) - with caplog.at_level(logging.DEBUG, logger='pyfive'): - with open(test_file, 'rb') as fh: + with caplog.at_level(logging.DEBUG, logger="pyfive"): + with open(test_file, "rb") as fh: try: do1 = DataObjects(fh, 0) except Exception: pass # Check if any pyfive diagnostic logs were captured - pyfive_logs = [r for r in caplog.records if 'pyfive' in r.getMessage()] + pyfive_logs = [r for r in caplog.records if "pyfive" in r.getMessage()] # Just verify we can create DataObjects without error @@ -84,16 +83,16 @@ def test_get_attributes_logs_file_handle_info(self, caplog, tmp_path): # to track whether attributes are being cached test_file = tmp_path / "test.hdf5" - with open(test_file, 'wb') as f: - f.write(b'\x89HDF\r\n\x1a\n') - f.write(b'\x00' * 100) + with open(test_file, "wb") as f: + f.write(b"\x89HDF\r\n\x1a\n") + f.write(b"\x00" * 100) - with caplog.at_level(logging.INFO, logger='pyfive'): - with open(test_file, 'rb') as fh: + with caplog.at_level(logging.INFO, logger="pyfive"): + with open(test_file, "rb") as fh: try: do = DataObjects(fh, 0) # Try to get attributes if possible - if hasattr(do, 'get_attributes'): + if hasattr(do, "get_attributes"): do.get_attributes() except Exception: pass @@ -101,8 +100,9 @@ def test_get_attributes_logs_file_handle_info(self, caplog, tmp_path): # Check for expected log message pattern # Format: "[pyfive] Obtained N attributes from offset Y (fh_id=X type=Y) in Z.XXXXs" logs_with_obtained = [ - r for r in caplog.records - if 'Obtained' in r.getMessage() and 'attributes' in r.getMessage() + r + for r in caplog.records + if "Obtained" in r.getMessage() and "attributes" in r.getMessage() ] # We may not have valid attributes, but the logging setup should be there @@ -110,10 +110,13 @@ def test_get_attributes_logs_file_handle_info(self, caplog, tmp_path): class TestS3FileHandleLifecycle: """Test S3 file handle lifecycle and reuse patterns.""" - @pytest.mark.parametrize("access_pattern", [ - "sequential", # Access each variable once - "repeated", # Access same variables multiple times - ]) + @pytest.mark.parametrize( + "access_pattern", + [ + "sequential", # Access each variable once + "repeated", # Access same variables multiple times + ], + ) def test_s3_file_handle_reuse(self, access_pattern): """ Test whether S3 file handles are reused across variable accesses. @@ -127,6 +130,7 @@ def test_s3_file_handle_reuse(self, access_pattern): class MockS3File: """Mock S3 file that tracks handle creation and reads.""" + _instance_counter = 0 def __init__(self, *args, **kwargs): @@ -136,11 +140,8 @@ def __init__(self, *args, **kwargs): file_handle_ids.append(self.id) def read(self, size=-1): - read_operations.append({ - 'handle_id': self.id, - 'size': size - }) - return b'\x00' * size if size > 0 else b'' + read_operations.append({"handle_id": self.id, "size": size}) + return b"\x00" * size if size > 0 else b"" def seek(self, offset): return offset @@ -149,7 +150,9 @@ def close(self): pass # Mock fsspec to use our MockS3File - with patch('fsspec.open', return_value=MagicMock(__enter__=lambda self: MockS3File())): + with patch( + "fsspec.open", return_value=MagicMock(__enter__=lambda self: MockS3File()) + ): if access_pattern == "sequential": # Each access creates a new file for i in range(3): @@ -169,8 +172,9 @@ def close(self): # Verify that in "repeated" pattern, the same handle is used if access_pattern == "repeated": - assert len(set(r['handle_id'] for r in read_operations)) == 1, \ + assert len(set(r["handle_id"] for r in read_operations)) == 1, ( "Expected same handle for repeated access" + ) class TestDuplicateAttributeReads: @@ -248,10 +252,10 @@ class TestFileHandleIdTracking: def test_dataobjects_init_logs_handle_id(self, caplog): """Test that DataObjects.__init__ logs file handle ID.""" - test_data = io.BytesIO(b'\x89HDF\r\n\x1a\n' + b'\x00' * 100) + test_data = io.BytesIO(b"\x89HDF\r\n\x1a\n" + b"\x00" * 100) test_data.seek(0) - with caplog.at_level(logging.DEBUG, logger='pyfive'): + with caplog.at_level(logging.DEBUG, logger="pyfive"): try: do = DataObjects(test_data, 0) except Exception: @@ -259,9 +263,13 @@ def test_dataobjects_init_logs_handle_id(self, caplog): # Look for the diagnostic log with fh_id, type, s3, offset # Format: "[pyfive] DataObjects init: fh_id= type= s3= offset=" - logs = [r.getMessage() for r in caplog.records if 'DataObjects init' in r.getMessage()] + logs = [ + r.getMessage() + for r in caplog.records + if "DataObjects init" in r.getMessage() + ] # Framework in place to capture this -if __name__ == '__main__': - pytest.main([__file__, '-v', '-s']) +if __name__ == "__main__": + pytest.main([__file__, "-v", "-s"]) From d698b1da896767333a8ddc8faa0cf95c835fb7cf Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 21 Apr 2026 12:38:22 +0000 Subject: [PATCH 18/24] Replace meaningless assert True with pytest.skip in test_duplicate_attribute_read_detection Agent-Logs-Url: https://github.com/NCAS-CMS/pyfive/sessions/e26762f6-6c10-4c60-9365-a21f50bb3fc3 Co-authored-by: bnlawrence <1792815+bnlawrence@users.noreply.github.com> --- tests/test_s3_caching.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/test_s3_caching.py b/tests/test_s3_caching.py index 30fdf553..b2a7863c 100644 --- a/tests/test_s3_caching.py +++ b/tests/test_s3_caching.py @@ -206,7 +206,10 @@ def test_duplicate_attribute_read_detection(self, caplog): # 3. If fh_id is DIFFERENT each time # → file handles being recreated unnecessarily - assert True, "See log output for file handle reuse patterns" + pytest.skip( + "This diagnostic test collected pyfive logs without asserting on them; " + "skip it until it is rewritten with concrete log expectations." + ) class TestReadAheadCacheStatistics: From c0a0b57bc15feb1131ad98d7fa61f5f5e1be8172 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Tue, 21 Apr 2026 13:57:05 +0100 Subject: [PATCH 19/24] Modificaitons in response to review --- pyfive/h5d.py | 24 ++++++---- tests/test_btree_parallel.py | 18 ++++---- tests/test_s3_caching.py | 88 ++++++++++++++++-------------------- 3 files changed, 62 insertions(+), 68 deletions(-) diff --git a/pyfive/h5d.py b/pyfive/h5d.py index 6554f458..17441d40 100644 --- a/pyfive/h5d.py +++ b/pyfive/h5d.py @@ -46,23 +46,24 @@ class ChunkRead: # Shared helpers # # ------------------------------------------------------------------ # - def set_parallelism(self, thread_count=5, cat_range_allowed=True, _btree_parallel=False): + def set_parallelism(self, thread_count=0, cat_range_allowed=True, btree_parallel=False): """ - Configure experimental chunk-read parallelism. + Configure chunk-read parallelism. ``thread_count`` controls POSIX threaded reads via ``os.pread``: - ``0`` disables threaded reads - ``>0`` enables threaded reads with that many workers - - Default 5 + - Default 4 ``cat_range_allowed`` enables fsspec bulk reads via ``cat_ranges`` for compatible non-posix file handles. Default True - ``parallel_btree`` enables parallel reads for b-tree nodes when building + ``btree_parallel`` enables parallel reads for b-tree nodes when building the chunk index. Default False. - This is a ``pyfive`` API extension, and is opt-in by default as it may not be suitable for all use cases. - It is recommended to enable it when working with remote files, but it may not be suitable for local files. + This is a ``pyfive`` API extension. It is recommended to enable it when working + with remote files, but it may not be suitable for local files. Hence defaults + are that cat_ranges is on (for remote files) and threads are off (for local files). """ if thread_count is None: @@ -72,8 +73,8 @@ def set_parallelism(self, thread_count=5, cat_range_allowed=True, _btree_paralle raise ValueError("thread_count must be >= 0") self._thread_count = thread_count - self._cat_range_allowed = bool(cat_range_allowed) - self._btree_parallel = bool(_btree_parallel) + self._cat_range_allowed = cat_range_allowed + self._btree_parallel = btree_parallel logger.info('Parallelism: thread_count=%d, cat_range_allowed=%s, btree_parallel=%s', self._thread_count, self._cat_range_allowed, self._btree_parallel) @@ -287,7 +288,6 @@ def __init__( self.shape = dataobject.shape self.rank = len(self.shape) self.chunks = dataobject.chunks - # Experimental chunk-read settings are opt-in by default. self.set_parallelism() # experimental code. We need to find out whether or not this @@ -618,7 +618,11 @@ def _build_index(self): self.__index_built = True def _make_btree_fetch_fn(self): - """Return fetch_fn(addresses, size) for b-tree leaf reads, or None.""" + """ + Return fetch_fn(addresses, size) for b-tree leaf reads, or None. + The default here is None (self._btree_parallel is None), + which means that b-tree nodes will be read serially via the file handle's read method. + """ if not self._btree_parallel: return None diff --git a/tests/test_btree_parallel.py b/tests/test_btree_parallel.py index 79df0110..470a9555 100644 --- a/tests/test_btree_parallel.py +++ b/tests/test_btree_parallel.py @@ -166,7 +166,7 @@ def test_leaf_node_size_uses_first_leaf_header(): def test_make_btree_fetch_fn_cat_ranges_case(): dsid = DatasetID.__new__(DatasetID) dsid.posix = False - dsid.set_parallelism(thread_count=0, cat_range_allowed=True, _btree_parallel=True) + dsid.set_parallelism(thread_count=0, cat_range_allowed=True, btree_parallel=True) fs = _DummyFS({10: b"abcd", 20: b"efgh"}) dsid._DatasetID__fh = _WrappedFH(fs, "bucket/file.h5") @@ -182,7 +182,7 @@ def test_make_btree_fetch_fn_cat_ranges_case(): def test_make_btree_fetch_fn_pread_case(tmp_path): dsid = DatasetID.__new__(DatasetID) dsid.posix = True - dsid.set_parallelism(thread_count=2, cat_range_allowed=False, _btree_parallel=True) + dsid.set_parallelism(thread_count=2, cat_range_allowed=False, btree_parallel=True) payload = b"abcdefghijklmnopqrstuvwxyz" fpath = tmp_path / "pread.bin" fpath.write_bytes(payload) @@ -198,7 +198,7 @@ def test_make_btree_fetch_fn_pread_case(tmp_path): def test_make_btree_fetch_fn_serial_case(): dsid = DatasetID.__new__(DatasetID) dsid.posix = True - dsid.set_parallelism(thread_count=0, cat_range_allowed=False, _btree_parallel=False) + dsid.set_parallelism(thread_count=0, cat_range_allowed=False, btree_parallel=False) fetch_fn = dsid._make_btree_fetch_fn() assert fetch_fn is None @@ -207,7 +207,7 @@ def test_make_btree_fetch_fn_serial_case(): def test_parallel_pread_matches_serial_results_for_chunked_dataset(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, _btree_parallel=False) + serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, btree_parallel=False) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -215,7 +215,7 @@ def test_parallel_pread_matches_serial_results_for_chunked_dataset(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: parallel = hfile["dataset1"] - parallel.id.set_parallelism(thread_count=2, cat_range_allowed=False, _btree_parallel=True) + parallel.id.set_parallelism(thread_count=2, cat_range_allowed=False, btree_parallel=True) parallel_data = parallel[:] parallel_chunk_info = [ parallel.id.get_chunk_info(i) for i in range(parallel.id.get_num_chunks()) @@ -228,7 +228,7 @@ def test_parallel_pread_matches_serial_results_for_chunked_dataset(): def test_parallel_fsspec_cat_ranges_matches_serial_results(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, _btree_parallel=False) + serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, btree_parallel=False) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -251,7 +251,7 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): with memfs.open(mem_path, "rb") as fh: with pyfive.File(fh) as hfile: ds = hfile["dataset1"] - ds.id.set_parallelism(thread_count=0, cat_range_allowed=True, _btree_parallel=True) + ds.id.set_parallelism(thread_count=0, cat_range_allowed=True, btree_parallel=True) fsspec_data = ds[:] fsspec_chunk_info = [ ds.id.get_chunk_info(i) for i in range(ds.id.get_num_chunks()) @@ -265,7 +265,7 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): def test_parallel_s3fs_cat_ranges_matches_serial_results(s3fs_s3): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, _btree_parallel=False) + serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, btree_parallel=False) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -292,7 +292,7 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): with s3fs_s3.open(s3_key, "rb") as fh: with pyfive.File(fh) as hfile: ds = hfile["dataset1"] - ds.id.set_parallelism(thread_count=0, cat_range_allowed=True, _btree_parallel=True) + ds.id.set_parallelism(thread_count=0, cat_range_allowed=True, btree_parallel=True) s3_data = ds[:] s3_chunk_info = [ ds.id.get_chunk_info(i) for i in range(ds.id.get_num_chunks()) diff --git a/tests/test_s3_caching.py b/tests/test_s3_caching.py index 324adff4..1ea77d08 100644 --- a/tests/test_s3_caching.py +++ b/tests/test_s3_caching.py @@ -14,11 +14,6 @@ import pytest -# Configure logging to capture pyfive diagnostic messages -logging.basicConfig(level=logging.DEBUG) -logger = logging.getLogger('pyfive') - - class TestFileHandleReuse: """Test whether file handles are reused across DataObjects instances.""" @@ -51,26 +46,24 @@ def test_dataobjects_tracks_file_handle_id(self, tmp_path): assert id(fh) == fh_id_first - def test_file_handle_identity_in_logging(self, caplog, tmp_path): - """Test that file handle ID is logged consistently.""" - from pyfive.dataobjects import DataObjects - - test_file = tmp_path / "test.hdf5" + def test_file_handle_identity_in_logging(self, caplog): + """Test that DataObjects.__init__ emits a diagnostic log recording the file handle ID.""" + from pyfive import File - with open(test_file, 'wb') as f: - f.write(b'\x89HDF\r\n\x1a\n') - f.write(b'\x00' * 100) + test_file = Path(__file__).parent / "compact.hdf5" with caplog.at_level(logging.DEBUG, logger='pyfive'): with open(test_file, 'rb') as fh: - try: - do1 = DataObjects(fh, 0) - except Exception: - pass + expected_fh_id = str(id(fh)) + f = File(fh) + list(f.keys()) + f.close() - # Check if any pyfive diagnostic logs were captured - pyfive_logs = [r for r in caplog.records if 'pyfive' in r.getMessage()] - # Just verify we can create DataObjects without error + init_logs = [r.getMessage() for r in caplog.records if 'DataObjects init' in r.getMessage()] + assert init_logs, "Expected at least one 'DataObjects init' log entry from DataObjects.__init__" + assert any(expected_fh_id in msg for msg in init_logs), ( + f"Expected fh_id={expected_fh_id} in log messages, got: {init_logs}" + ) class TestAttributeReadCaching: @@ -148,24 +141,16 @@ def seek(self, offset): def close(self): pass - # Mock fsspec to use our MockS3File - with patch('fsspec.open', return_value=MagicMock(__enter__=lambda self: MockS3File())): - if access_pattern == "sequential": - # Each access creates a new file - for i in range(3): - try: - f = MockS3File() - f.read(100) - except Exception: - pass - else: # repeated - # Same file accessed multiple times + if access_pattern == "sequential": + # Each access creates a new file handle + for i in range(3): f = MockS3File() - for i in range(3): - try: - f.read(100) - except Exception: - pass + f.read(100) + else: # repeated + # Same file handle accessed multiple times + f = MockS3File() + for i in range(3): + f.read(100) # Verify that in "repeated" pattern, the same handle is used if access_pattern == "repeated": @@ -185,24 +170,29 @@ class TestDuplicateAttributeReads: or cached reads of the same data. """ + @pytest.mark.skip( + reason=( + "Not yet automated: needs a real or fully-mocked HDF5 file that triggers " + "both collect_dimensions_from_root() and dump_header() so that duplicate " + "DataObjects reads of the same offset can be asserted against captured logs " + "or a mock read counter." + ) + ) def test_duplicate_attribute_read_detection(self, caplog): """ Test framework to detect duplicate attribute reads. Logs should show if same file handle is reading same offset twice. + + Expected behavior to verify: + 1. If fh_id is SAME both times and offset is SAME but timing is DIFFERENT + → fsspec cache NOT working properly (both are misses) + 2. If fh_id is SAME, offset is SAME, timing is FAST second time + → fsspec cache IS working (first miss, second hit) + 3. If fh_id is DIFFERENT each time + → file handles being recreated unnecessarily """ - # This is a framework test - in practice, run with: - # python -m pytest tests/test_s3_caching.py::TestDuplicateAttributeReads::test_duplicate_attribute_read_detection -v -s --log-cli-level=DEBUG - - # Expected behavior to verify: - # 1. If fh_id is SAME both times and offset is SAME but timing is DIFFERENT - # → fsspec cache NOT working properly (both are misses) - # 2. If fh_id is SAME, offset is SAME, timing is FAST second time - # → fsspec cache IS working (first miss, second hit) - # 3. If fh_id is DIFFERENT each time - # → file handles being recreated unnecessarily - - assert True, "See log output for file handle reuse patterns" + raise NotImplementedError("Replace with assertions against captured logs or mock read counts") class TestReadAheadCacheStatistics: From d0ab7182cb5bbed75709d164230e8ab08ae19ab2 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Tue, 21 Apr 2026 14:07:42 +0100 Subject: [PATCH 20/24] Fixing two blank lines at the end, because that seems to be causing pre-commit fail???? --- tests/test_btree_parallel.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_btree_parallel.py b/tests/test_btree_parallel.py index 470a9555..d9276f0e 100644 --- a/tests/test_btree_parallel.py +++ b/tests/test_btree_parallel.py @@ -301,4 +301,3 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): assert calls, "Expected s3fs cat_ranges to be used for leaf-node reads" assert_array_equal(s3_data, serial_data) assert s3_chunk_info == serial_chunk_info - From 4f7878c275b781a5511e60106e7b1da0c06a4e58 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Tue, 21 Apr 2026 14:19:19 +0100 Subject: [PATCH 21/24] pre-commit magic that V made me do --- pyfive/h5d.py | 28 +++++++++---- tests/test_btree_parallel.py | 24 ++++++++--- tests/test_s3_caching.py | 80 +++++++++++++++++++++--------------- 3 files changed, 85 insertions(+), 47 deletions(-) diff --git a/pyfive/h5d.py b/pyfive/h5d.py index 17441d40..27d2aec9 100644 --- a/pyfive/h5d.py +++ b/pyfive/h5d.py @@ -26,7 +26,6 @@ ChunkIndex = namedtuple("ChunkIndex", "chunk_address chunk_dims") - class ChunkRead: """ Mixin providing parallel and bulk chunk-reading strategies. @@ -46,7 +45,9 @@ class ChunkRead: # Shared helpers # # ------------------------------------------------------------------ # - def set_parallelism(self, thread_count=0, cat_range_allowed=True, btree_parallel=False): + def set_parallelism( + self, thread_count=0, cat_range_allowed=True, btree_parallel=False + ): """ Configure chunk-read parallelism. @@ -75,9 +76,12 @@ def set_parallelism(self, thread_count=0, cat_range_allowed=True, btree_parallel self._thread_count = thread_count self._cat_range_allowed = cat_range_allowed self._btree_parallel = btree_parallel - logger.info('Parallelism: thread_count=%d, cat_range_allowed=%s, btree_parallel=%s', - self._thread_count, self._cat_range_allowed, self._btree_parallel) - + logger.info( + "Parallelism: thread_count=%d, cat_range_allowed=%s, btree_parallel=%s", + self._thread_count, + self._cat_range_allowed, + self._btree_parallel, + ) def _get_required_chunks(self, indexer): """ @@ -121,7 +125,9 @@ def _select_chunks(self, indexer, out, dtype): fh = self._fh actual_fh = getattr(fh, "fh", fh) # support wrapped file-like objects if hasattr(actual_fh, "fs") and hasattr(actual_fh.fs, "cat_ranges"): - logger.info(f"[pyfive] chunk read strategy: fsspec_cat_ranges ({len(chunks)} chunks)") + logger.info( + f"[pyfive] chunk read strategy: fsspec_cat_ranges ({len(chunks)} chunks)" + ) self._read_bulk_fsspec(fh, chunks, out, dtype) return @@ -182,7 +188,9 @@ def _read_one(item): finally: fh.close() - logger.info('pyfive thread pool read completed using %d threads', self._thread_count) + logger.info( + "pyfive thread pool read completed using %d threads", self._thread_count + ) for chunk_sel, out_sel, filter_mask, chunk_buffer in results: out[out_sel] = self._decode_chunk(chunk_buffer, filter_mask, dtype)[ @@ -654,7 +662,11 @@ def fetch_pread(addresses, size): fd = fh_local.fileno() try: with ThreadPoolExecutor(max_workers=thread_count) as executor: - return list(executor.map(lambda addr: os.pread(fd, size, addr), addresses)) + return list( + executor.map( + lambda addr: os.pread(fd, size, addr), addresses + ) + ) finally: fh_local.close() diff --git a/tests/test_btree_parallel.py b/tests/test_btree_parallel.py index d9276f0e..1be560be 100644 --- a/tests/test_btree_parallel.py +++ b/tests/test_btree_parallel.py @@ -207,7 +207,9 @@ def test_make_btree_fetch_fn_serial_case(): def test_parallel_pread_matches_serial_results_for_chunked_dataset(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, btree_parallel=False) + serial.id.set_parallelism( + thread_count=0, cat_range_allowed=False, btree_parallel=False + ) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -215,7 +217,9 @@ def test_parallel_pread_matches_serial_results_for_chunked_dataset(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: parallel = hfile["dataset1"] - parallel.id.set_parallelism(thread_count=2, cat_range_allowed=False, btree_parallel=True) + parallel.id.set_parallelism( + thread_count=2, cat_range_allowed=False, btree_parallel=True + ) parallel_data = parallel[:] parallel_chunk_info = [ parallel.id.get_chunk_info(i) for i in range(parallel.id.get_num_chunks()) @@ -228,7 +232,9 @@ def test_parallel_pread_matches_serial_results_for_chunked_dataset(): def test_parallel_fsspec_cat_ranges_matches_serial_results(): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, btree_parallel=False) + serial.id.set_parallelism( + thread_count=0, cat_range_allowed=False, btree_parallel=False + ) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -251,7 +257,9 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): with memfs.open(mem_path, "rb") as fh: with pyfive.File(fh) as hfile: ds = hfile["dataset1"] - ds.id.set_parallelism(thread_count=0, cat_range_allowed=True, btree_parallel=True) + ds.id.set_parallelism( + thread_count=0, cat_range_allowed=True, btree_parallel=True + ) fsspec_data = ds[:] fsspec_chunk_info = [ ds.id.get_chunk_info(i) for i in range(ds.id.get_num_chunks()) @@ -265,7 +273,9 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): def test_parallel_s3fs_cat_ranges_matches_serial_results(s3fs_s3): with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile: serial = hfile["dataset1"] - serial.id.set_parallelism(thread_count=0, cat_range_allowed=False, btree_parallel=False) + serial.id.set_parallelism( + thread_count=0, cat_range_allowed=False, btree_parallel=False + ) serial_data = serial[:] serial_chunk_info = [ serial.id.get_chunk_info(i) for i in range(serial.id.get_num_chunks()) @@ -292,7 +302,9 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs): with s3fs_s3.open(s3_key, "rb") as fh: with pyfive.File(fh) as hfile: ds = hfile["dataset1"] - ds.id.set_parallelism(thread_count=0, cat_range_allowed=True, btree_parallel=True) + ds.id.set_parallelism( + thread_count=0, cat_range_allowed=True, btree_parallel=True + ) s3_data = ds[:] s3_chunk_info = [ ds.id.get_chunk_info(i) for i in range(ds.id.get_num_chunks()) diff --git a/tests/test_s3_caching.py b/tests/test_s3_caching.py index 1ea77d08..554d2e1d 100644 --- a/tests/test_s3_caching.py +++ b/tests/test_s3_caching.py @@ -25,14 +25,14 @@ def test_dataobjects_tracks_file_handle_id(self, tmp_path): test_file = tmp_path / "test.hdf5" # Create a minimal HDF5 file by writing bytes - with open(test_file, 'wb') as f: + with open(test_file, "wb") as f: # HDF5 signature - f.write(b'\x89HDF\r\n\x1a\n') + f.write(b"\x89HDF\r\n\x1a\n") # Minimal superblock (rest can be zeros for this test) - f.write(b'\x00' * 100) + f.write(b"\x00" * 100) # Open file and create DataObjects instances - with open(test_file, 'rb') as fh: + with open(test_file, "rb") as fh: fh_id_first = id(fh) # Create first DataObjects instance @@ -45,22 +45,27 @@ def test_dataobjects_tracks_file_handle_id(self, tmp_path): # The file handle should be the same object assert id(fh) == fh_id_first - def test_file_handle_identity_in_logging(self, caplog): """Test that DataObjects.__init__ emits a diagnostic log recording the file handle ID.""" from pyfive import File test_file = Path(__file__).parent / "compact.hdf5" - with caplog.at_level(logging.DEBUG, logger='pyfive'): - with open(test_file, 'rb') as fh: + with caplog.at_level(logging.DEBUG, logger="pyfive"): + with open(test_file, "rb") as fh: expected_fh_id = str(id(fh)) f = File(fh) list(f.keys()) f.close() - init_logs = [r.getMessage() for r in caplog.records if 'DataObjects init' in r.getMessage()] - assert init_logs, "Expected at least one 'DataObjects init' log entry from DataObjects.__init__" + init_logs = [ + r.getMessage() + for r in caplog.records + if "DataObjects init" in r.getMessage() + ] + assert init_logs, ( + "Expected at least one 'DataObjects init' log entry from DataObjects.__init__" + ) assert any(expected_fh_id in msg for msg in init_logs), ( f"Expected fh_id={expected_fh_id} in log messages, got: {init_logs}" ) @@ -77,16 +82,16 @@ def test_get_attributes_logs_file_handle_info(self, caplog, tmp_path): # to track whether attributes are being cached test_file = tmp_path / "test.hdf5" - with open(test_file, 'wb') as f: - f.write(b'\x89HDF\r\n\x1a\n') - f.write(b'\x00' * 100) + with open(test_file, "wb") as f: + f.write(b"\x89HDF\r\n\x1a\n") + f.write(b"\x00" * 100) - with caplog.at_level(logging.INFO, logger='pyfive'): - with open(test_file, 'rb') as fh: + with caplog.at_level(logging.INFO, logger="pyfive"): + with open(test_file, "rb") as fh: try: do = DataObjects(fh, 0) # Try to get attributes if possible - if hasattr(do, 'get_attributes'): + if hasattr(do, "get_attributes"): do.get_attributes() except Exception: pass @@ -94,8 +99,9 @@ def test_get_attributes_logs_file_handle_info(self, caplog, tmp_path): # Check for expected log message pattern # Format: "[pyfive] Obtained N attributes from offset Y (fh_id=X type=Y) in Z.XXXXs" logs_with_obtained = [ - r for r in caplog.records - if 'Obtained' in r.getMessage() and 'attributes' in r.getMessage() + r + for r in caplog.records + if "Obtained" in r.getMessage() and "attributes" in r.getMessage() ] # We may not have valid attributes, but the logging setup should be there @@ -103,10 +109,13 @@ def test_get_attributes_logs_file_handle_info(self, caplog, tmp_path): class TestS3FileHandleLifecycle: """Test S3 file handle lifecycle and reuse patterns.""" - @pytest.mark.parametrize("access_pattern", [ - "sequential", # Access each variable once - "repeated", # Access same variables multiple times - ]) + @pytest.mark.parametrize( + "access_pattern", + [ + "sequential", # Access each variable once + "repeated", # Access same variables multiple times + ], + ) def test_s3_file_handle_reuse(self, access_pattern): """ Test whether S3 file handles are reused across variable accesses. @@ -120,6 +129,7 @@ def test_s3_file_handle_reuse(self, access_pattern): class MockS3File: """Mock S3 file that tracks handle creation and reads.""" + _instance_counter = 0 def __init__(self, *args, **kwargs): @@ -129,11 +139,8 @@ def __init__(self, *args, **kwargs): file_handle_ids.append(self.id) def read(self, size=-1): - read_operations.append({ - 'handle_id': self.id, - 'size': size - }) - return b'\x00' * size if size > 0 else b'' + read_operations.append({"handle_id": self.id, "size": size}) + return b"\x00" * size if size > 0 else b"" def seek(self, offset): return offset @@ -154,8 +161,9 @@ def close(self): # Verify that in "repeated" pattern, the same handle is used if access_pattern == "repeated": - assert len(set(r['handle_id'] for r in read_operations)) == 1, \ + assert len(set(r["handle_id"] for r in read_operations)) == 1, ( "Expected same handle for repeated access" + ) class TestDuplicateAttributeReads: @@ -192,7 +200,9 @@ def test_duplicate_attribute_read_detection(self, caplog): 3. If fh_id is DIFFERENT each time → file handles being recreated unnecessarily """ - raise NotImplementedError("Replace with assertions against captured logs or mock read counts") + raise NotImplementedError( + "Replace with assertions against captured logs or mock read counts" + ) class TestReadAheadCacheStatistics: @@ -238,10 +248,10 @@ class TestFileHandleIdTracking: def test_dataobjects_init_logs_handle_id(self, caplog): """Test that DataObjects.__init__ logs file handle ID.""" - test_data = io.BytesIO(b'\x89HDF\r\n\x1a\n' + b'\x00' * 100) + test_data = io.BytesIO(b"\x89HDF\r\n\x1a\n" + b"\x00" * 100) test_data.seek(0) - with caplog.at_level(logging.DEBUG, logger='pyfive'): + with caplog.at_level(logging.DEBUG, logger="pyfive"): try: do = DataObjects(test_data, 0) except Exception: @@ -249,9 +259,13 @@ def test_dataobjects_init_logs_handle_id(self, caplog): # Look for the diagnostic log with fh_id, type, s3, offset # Format: "[pyfive] DataObjects init: fh_id= type= s3= offset=" - logs = [r.getMessage() for r in caplog.records if 'DataObjects init' in r.getMessage()] + logs = [ + r.getMessage() + for r in caplog.records + if "DataObjects init" in r.getMessage() + ] # Framework in place to capture this -if __name__ == '__main__': - pytest.main([__file__, '-v', '-s']) +if __name__ == "__main__": + pytest.main([__file__, "-v", "-s"]) From b134783704297fc1a2a83f9a59c710b02f479fe7 Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Tue, 21 Apr 2026 14:33:39 +0100 Subject: [PATCH 22/24] I'm now confused about which files need to be in the repo and which don't --- tests/compact.hdf5 | Bin 0 -> 1416 bytes tests/opaque_datetime.hdf5 | Bin 0 -> 6228 bytes tests/opaque_fixed.hdf5 | Bin 0 -> 2240 bytes 3 files changed, 0 insertions(+), 0 deletions(-) create mode 100644 tests/compact.hdf5 create mode 100644 tests/opaque_datetime.hdf5 create mode 100644 tests/opaque_fixed.hdf5 diff --git a/tests/compact.hdf5 b/tests/compact.hdf5 new file mode 100644 index 0000000000000000000000000000000000000000..815391be2809a061fcb732ddea84400804e5fca9 GIT binary patch literal 1416 zcmeD5aB<`1lHy_j0S*oZ76t(@6Gr@pf(}-Q2#gPtPk=HQp>zk7Ucm%mFfxE31A_!q zTo7tLy1I}cS62q0N|^aD8mf)Kfd#_ifC-G!BPs+uTpa^I9*%(e8kR~=K+_p4Fp~3g z3lft{z-bO7<^hu+lV)UK1DhoPP2h~sl+4Ho5r8I3V7`F>W?=dT$p`=wDnv{HEWyga z0TyRsWCCmCfSLnS5Ap^x5W~z70;&fAkm9UePcj%7CV=^ajR^MhcLC*ah-(=b6hNLB I1tf<606KaxfB*mh literal 0 HcmV?d00001 diff --git a/tests/opaque_datetime.hdf5 b/tests/opaque_datetime.hdf5 new file mode 100644 index 0000000000000000000000000000000000000000..54e6db167b587199684a15c582b0542d3eed8ab0 GIT binary patch literal 6228 zcmeHK%}xR_5S}hzHi!`qCg^D&fuk`VFoNzuexjIYj6a*O5e~AttWlppym|5&y!sZt z22Y}$b|yI?G`3{Oj%dRevr;+2+>V& zdlz1Gf+S3O*WrNUgCy?tuP`2%P!xB1{UCni2A4eLHM4s|qrFSnxRU zDMu?)e7GZ)V`z`uln>+Wyq>gRkE zO`=7eo&9U@Ok&gP%csIBFacy=CdTtMe0_g1_Y#a2&oT2#Vbn#7W$lx4&>tV;m5XJ{ z^QgnUo#wvLBMMVDt4)i;ang;$P+PvP`PxYCBb+w-kK3q^=(N2pr@nZ?fG{8o2m``^ eFdz&F1HynXAPfit!hkR^8wT*bk8gW?*M9*W$5EXC literal 0 HcmV?d00001 diff --git a/tests/opaque_fixed.hdf5 b/tests/opaque_fixed.hdf5 new file mode 100644 index 0000000000000000000000000000000000000000..6b36628cfa3fbcbec7eb8f8ea59265a98b5de0be GIT binary patch literal 2240 zcmeD5aB<`1lHy_j0S*oZ76t(@6Gr@pf&&~75f~pPp8#brLg@}Dy@CnCU}OM61_lYJ zxFFPgbaf#?uC5F~l`!*RG*lad0Skl$bp}j$lpY}=;Nj{R0P<=C)W5LwbOM^rV8M`I zkXTrn8lRF_k_Z-prAr5x0upIP23BzTg%F?=3bC1y5n?hk)I0>u0ahvs6n6k(Rt5;4 ziHQlUfdi_anGq-f1_vM%oSXq?4Ms+=pT7$zN5k@=0?gxs(Xpd$8V!Nb5WpJ(8L2rr y`3mLvML8*W)!~w4WMXDXE-fy}&rQnAODw8{7reMMpo@bHU}aCFfIPqe literal 0 HcmV?d00001 From 9c85ad3e1b849a20eed867157918f60dd6bd3ded Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Tue, 28 Apr 2026 15:55:56 +0100 Subject: [PATCH 23/24] Documentation addition and replacement of OrderedDicts with Dicts as requested in https://github.com/NCAS-CMS/pyfive/pull/218#pullrequestreview-4188133904 --- pyfive/btree.py | 18 +++++++++--------- pyfive/high_level.py | 5 +++++ pyfive/utilities.py | 3 +++ 3 files changed, 17 insertions(+), 9 deletions(-) diff --git a/pyfive/btree.py b/pyfive/btree.py index b3e1adb3..586bf241 100644 --- a/pyfive/btree.py +++ b/pyfive/btree.py @@ -1,6 +1,6 @@ """HDF5 B-Trees and contents.""" -from collections import OrderedDict +from typing import Dict import struct import zlib @@ -66,7 +66,7 @@ class BTreeV1(AbstractBTree): """ # III.A.1. Disk Format: Level 1A1 - Version 1 B-trees - B_LINK_NODE = OrderedDict( + B_LINK_NODE = Dict( ( ("signature", "4s"), ("node_type", "B"), @@ -163,7 +163,7 @@ def _read_children(self): header_size = struct.calcsize("<" + "".join(self.B_LINK_NODE.values())) header_buffers = self._fetch_fn(leaf_addresses, header_size) - size_to_addresses = OrderedDict() + size_to_addresses = Dict() entry_size = 8 + self.dims * 8 + 8 for addr, header in zip(leaf_addresses, header_buffers): entries_used = struct.unpack_from(" None: """initalize.""" self.parent = parent diff --git a/pyfive/utilities.py b/pyfive/utilities.py index 3089f92d..54b8b4e6 100644 --- a/pyfive/utilities.py +++ b/pyfive/utilities.py @@ -79,13 +79,16 @@ def closed(self) -> bool: @property def fs(self): + """Return underlying filesystem if available.""" return getattr(self.fh, "fs", None) @property def path(self): + """Return underlying file path if available.""" return getattr(self.fh, "path", None) def __getattr__(self, name: str): + """Return attributes from underlying file handle for any attributes not defined on wrapper.""" return getattr(self.fh, name) def _ensure_buffer(self): From c56e74ce756cccddf2fbe0b178223f9a834b20cf Mon Sep 17 00:00:00 2001 From: Bryan Lawrence Date: Tue, 28 Apr 2026 16:01:02 +0100 Subject: [PATCH 24/24] apparently I am stupid, Dict instead of dict, and no testing. Bad. --- pyfive/btree.py | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/pyfive/btree.py b/pyfive/btree.py index 586bf241..5a3636fb 100644 --- a/pyfive/btree.py +++ b/pyfive/btree.py @@ -1,6 +1,5 @@ """HDF5 B-Trees and contents.""" -from typing import Dict import struct import zlib @@ -66,7 +65,7 @@ class BTreeV1(AbstractBTree): """ # III.A.1. Disk Format: Level 1A1 - Version 1 B-trees - B_LINK_NODE = Dict( + B_LINK_NODE = dict( ( ("signature", "4s"), ("node_type", "B"), @@ -163,7 +162,7 @@ def _read_children(self): header_size = struct.calcsize("<" + "".join(self.B_LINK_NODE.values())) header_buffers = self._fetch_fn(leaf_addresses, header_size) - size_to_addresses = Dict() + size_to_addresses = dict() entry_size = 8 + self.dims * 8 + 8 for addr, header in zip(leaf_addresses, header_buffers): entries_used = struct.unpack_from("