From 608de9ac9b2b9d35beb83e77c920c3e52483d6f0 Mon Sep 17 00:00:00 2001 From: Kenya Otsuka Date: Fri, 18 Sep 2026 04:51:53 +0000 Subject: [PATCH 01/11] feat(dataform): add chunk-shape policy for chunked feature storage HDF5 reads whole chunks even when only a few of their elements are selected, so the chunk shape is what decides the cost of a partial read. Add choose_chunk_shape(), which derives a chunk shape from a byte budget (1 MiB by default, per h5py guidance) instead of hard-coding an extent along any one axis. Both axes callers actually slice are chunked: axis 0 (samples) and axis 1 (channels/units). Trailing spatial axes are kept whole -- they are small and never sliced, and splitting them would only shrink the chunks. The budget is split evenly between the two chunked axes, with each absorbing the headroom the other leaves, so a short sample axis does not force a narrow feature chunk. --- bdpy/dataform/_feature_chunking.py | 108 ++++++++++++++++++++++ tests/dataform/test_feature_chunking.py | 117 ++++++++++++++++++++++++ 2 files changed, 225 insertions(+) create mode 100644 bdpy/dataform/_feature_chunking.py create mode 100644 tests/dataform/test_feature_chunking.py diff --git a/bdpy/dataform/_feature_chunking.py b/bdpy/dataform/_feature_chunking.py new file mode 100644 index 0000000..44b1b45 --- /dev/null +++ b/bdpy/dataform/_feature_chunking.py @@ -0,0 +1,108 @@ +"""Chunk-shape policy for chunked feature storage. + +HDF5 reads whole chunks even when only a few elements of a chunk are selected, +so the chunk shape - not the file format alone - is what decides the cost of a +partial read. This module turns an array shape and dtype into a chunk shape +under a byte budget, rather than hard-coding an extent along any one axis. + +This file is a part of BdPy. +""" + +import math +from typing import Sequence, Tuple + +import numpy as np + +__all__ = ["DEFAULT_TARGET_CHUNK_BYTES", "choose_chunk_shape"] + +#: Default per-chunk byte budget. h5py's own guidance puts a useful chunk +#: somewhere between 10 KiB and 1 MiB; features are large and read in slabs, so +#: we sit at the top of that range. +DEFAULT_TARGET_CHUNK_BYTES = 1 << 20 + + +def _prod(values: Sequence[int]) -> int: + out = 1 + for v in values: + out *= int(v) + return out + + +def choose_chunk_shape( + shape: Sequence[int], + dtype: np.dtype, + target_bytes: int = DEFAULT_TARGET_CHUNK_BYTES, + n_samples_known: bool = True, +) -> Tuple[int, ...]: + """Choose an HDF5 chunk shape for a feature array under a byte budget. + + The two axes callers actually slice are axis 0 (samples) and axis 1 (the + outermost feature axis, typically channels or units), so both are chunked. + Trailing axes - the spatial axes of a convolutional feature map, say + ``(13, 13)`` - are kept whole: they are small and are never sliced in + practice, and splitting them would only shrink the chunks. + + Within the budget the two chunked axes are split roughly evenly in element + count, so that neither a sample-wise read nor a channel-wise read degenerates + into reading the entire dataset. + + Parameters + ---------- + shape : sequence of int + Full array shape, ``(n_samples, *feature_shape)``. + dtype : numpy.dtype + Array dtype; only its itemsize matters. + target_bytes : int, optional + Upper bound on the size of one chunk in bytes + (default: ``DEFAULT_TARGET_CHUNK_BYTES``). + n_samples_known : bool, optional + Whether ``shape[0]`` is the final number of samples. Pass ``False`` when + writing to a resizable dataset whose sample count is not yet known, so + that the sample-axis extent is not capped by the current size. + + Returns + ------- + tuple of int + Chunk shape, same length as ``shape``. Every element is at least 1 and, + when ``n_samples_known`` is True, at most the corresponding entry of + ``shape``. + + Examples + -------- + >>> choose_chunk_shape((1200, 1000), np.dtype(np.float32)) + (512, 512) + >>> choose_chunk_shape((1200, 256, 13, 13), np.dtype(np.float32)) + (39, 39, 13, 13) + >>> choose_chunk_shape((50, 1000), np.dtype(np.float32)) + (50, 1000) + """ + shape = tuple(int(s) for s in shape) + if len(shape) < 2: + raise ValueError( + "Feature arrays need at least a sample axis and a feature axis; " + "got shape {}".format(shape) + ) + if target_bytes < 1: + raise ValueError("target_bytes must be positive, got {}".format(target_bytes)) + + itemsize = np.dtype(dtype).itemsize + + # The whole array fits in one chunk: nothing to gain from splitting it. + if n_samples_known and _prod(shape) * itemsize <= target_bytes: + return shape + + # Bytes taken by one (sample, feature-0) cell, i.e. one element of the two + # chunked axes with the trailing axes kept whole. + inner = max(1, _prod(shape[2:])) * itemsize + budget_cells = max(1, target_bytes // inner) + + # Split the budget evenly between the two chunked axes, then let each axis + # absorb the headroom the other did not need: a short sample axis should not + # force a narrow feature chunk, and vice versa. + c1 = min(shape[1], max(1, math.isqrt(budget_cells))) + c0 = max(1, budget_cells // c1) + if n_samples_known: + c0 = min(shape[0], c0) + c1 = min(shape[1], max(c1, budget_cells // c0)) + + return (c0, c1, *shape[2:]) diff --git a/tests/dataform/test_feature_chunking.py b/tests/dataform/test_feature_chunking.py new file mode 100644 index 0000000..579f70c --- /dev/null +++ b/tests/dataform/test_feature_chunking.py @@ -0,0 +1,117 @@ +import unittest + +import numpy as np + +from bdpy.dataform._feature_chunking import ( + DEFAULT_TARGET_CHUNK_BYTES, + choose_chunk_shape, +) + + +def _nbytes(chunk, dtype): + return int(np.prod(chunk)) * np.dtype(dtype).itemsize + + +class TestChooseChunkShape(unittest.TestCase): + """Chunk shape decides partial-read cost, so pin the policy down.""" + + shapes = [ + (1200, 1000), + (1200, 256, 13, 13), + (50, 1000), + (1, 256, 13, 13), + (3, 4096, 7, 7), + (1000000, 1), + (2, 1000000), + (17, 63, 5), + ] + + def test_invariants(self): + for dtype in (np.float32, np.float64, np.int16): + for shape in self.shapes: + chunk = choose_chunk_shape(shape, np.dtype(dtype)) + with self.subTest(shape=shape, dtype=dtype): + self.assertEqual(len(chunk), len(shape)) + # Never zero, never larger than the dataset. + for c, s in zip(chunk, shape): + self.assertGreaterEqual(c, 1) + self.assertLessEqual(c, s) + # Trailing axes are kept whole. + self.assertEqual(chunk[2:], tuple(shape[2:])) + + def test_chunk_stays_within_budget(self): + dtype = np.dtype(np.float32) + for shape in self.shapes: + chunk = choose_chunk_shape(shape, dtype) + with self.subTest(shape=shape): + if _nbytes(shape, dtype) <= DEFAULT_TARGET_CHUNK_BYTES: + # Small enough to store as a single chunk. + self.assertEqual(chunk, tuple(shape)) + else: + self.assertLessEqual( + _nbytes(chunk, dtype), DEFAULT_TARGET_CHUNK_BYTES + ) + + def test_budget_is_actually_used(self): + # A chunk far below budget means needless read amplification: every + # oversized array should fill at least a quarter of the budget. + dtype = np.dtype(np.float32) + for shape in self.shapes: + if _nbytes(shape, dtype) <= DEFAULT_TARGET_CHUNK_BYTES: + continue + chunk = choose_chunk_shape(shape, dtype) + with self.subTest(shape=shape): + self.assertGreater( + _nbytes(chunk, dtype), DEFAULT_TARGET_CHUNK_BYTES // 4 + ) + + def test_both_sliceable_axes_are_chunked(self): + # The point of #144: a channel slice must not have to read every channel. + shape = (1200, 256, 13, 13) + chunk = choose_chunk_shape(shape, np.dtype(np.float32)) + self.assertLess(chunk[0], shape[0]) + self.assertLess(chunk[1], shape[1]) + + def test_whole_array_below_budget_is_one_chunk(self): + shape = (50, 1000) + self.assertEqual( + choose_chunk_shape(shape, np.dtype(np.float32)), shape + ) + + def test_single_sample(self): + chunk = choose_chunk_shape((1, 4096, 7, 7), np.dtype(np.float32)) + self.assertEqual(chunk[0], 1) + + def test_target_bytes_is_respected(self): + dtype = np.dtype(np.float32) + shape = (1200, 256, 13, 13) + small = choose_chunk_shape(shape, dtype, target_bytes=64 * 1024) + large = choose_chunk_shape(shape, dtype, target_bytes=4 * 1024 * 1024) + self.assertLessEqual(_nbytes(small, dtype), 64 * 1024) + self.assertLessEqual(_nbytes(large, dtype), 4 * 1024 * 1024) + self.assertLess(_nbytes(small, dtype), _nbytes(large, dtype)) + + def test_unknown_sample_count_is_not_capped(self): + # Resizable datasets start empty; the sample extent must come from the + # budget, not from the current (zero) size. + chunk = choose_chunk_shape( + (0, 256, 13, 13), np.dtype(np.float32), n_samples_known=False + ) + self.assertGreater(chunk[0], 1) + + def test_deterministic(self): + shape = (1200, 256, 13, 13) + dtype = np.dtype(np.float32) + self.assertEqual( + choose_chunk_shape(shape, dtype), choose_chunk_shape(shape, dtype) + ) + + def test_rejects_bad_input(self): + with self.assertRaises(ValueError): + choose_chunk_shape((100,), np.dtype(np.float32)) + with self.assertRaises(ValueError): + choose_chunk_shape((100, 10), np.dtype(np.float32), target_bytes=0) + + +if __name__ == "__main__": + unittest.main() From 74230ad2b0ecf1a0db79a9ed65ada8652987ca9a Mon Sep 17 00:00:00 2001 From: Kenya Otsuka Date: Fri, 18 Sep 2026 04:55:54 +0000 Subject: [PATCH 02/11] feat(dataform): read chunked HDF5 features through a FeatureStore backend Features was welded to one on-disk layout: a directory per layer holding one .mat file per stimulus. That layout has a file boundary only on the sample axis, so reading a few channels still costs a full read of every stimulus (issue #144). Introduce a FeatureStore interface so the layout is a backend detail: - MatFeatureStore carries today's behaviour unchanged (per-label paths, key 'feat', _mat_v73.loadmat_key, the multiprocessing fan-out). - HDF5FeatureStore reads .h5 holding /features and /labels, with /features explicitly chunked, and pushes feature-axis slices down to h5py so only the covered chunks are read. Root attributes bdpy_format and bdpy_format_version identify the file and are validated on open. Features holds one store per dpath and dispatches reads to whichever store owns each label, so a mix of layouts across dpath entries works and a future backend needs no change here. The format is auto-detected per directory and can be forced with format='mat'/'hdf5'. New read API, identical on both backends: features.get(layer, label=..., feature_slice=np.s_[128:256]) features.iter_chunks(layer, axis=1) feature_slice is a logical slice rather than the get_chunk() of the issue, so the HDF5 chunk concept stays out of the public API; on the legacy backend it loads in full and slices, on HDF5 it is a real partial read. iter_chunks yields (slice, block) so callers can place results back without tracking offsets, and defaults to the on-disk chunk extent so every element is read exactly once. Existing behaviour is untouched: labels, index, layers, feature_index, get, get_features, statistic, the single-slot layer cache and statistics.pkl all behave exactly as before, including caller-order row selection. Sliced reads deliberately bypass the layer cache, which exists to hold a whole layer. --- bdpy/dataform/_feature_store.py | 580 ++++++++++++++++++++++++++++++++ bdpy/dataform/features.py | 295 +++++++++++----- 2 files changed, 788 insertions(+), 87 deletions(-) create mode 100644 bdpy/dataform/_feature_store.py diff --git a/bdpy/dataform/_feature_store.py b/bdpy/dataform/_feature_store.py new file mode 100644 index 0000000..38e2112 --- /dev/null +++ b/bdpy/dataform/_feature_store.py @@ -0,0 +1,580 @@ +"""Storage backends for DNN feature collections. + +``Features`` used to be welded to one on-disk layout: a directory per layer +holding one ``.mat`` file per stimulus. That layout has a file boundary only on +the sample axis, so reading a few channels still costs a full read of every +stimulus. This module puts a backend interface between ``Features`` and the +bytes, so the legacy layout and the chunked HDF5 layout are interchangeable and +a future backend needs no change to ``Features`` itself. + +Every backend answers the same question -- "give me these labels, sliced this +way along the feature axes" -- and differs only in how much it has to read to do +it. The legacy backend loads whole stimulus files and slices in memory; the HDF5 +backend pushes the slice down to h5py and reads only the chunks it covers. + +This file is a part of BdPy. +""" + +import glob +import os +from abc import ABC, abstractmethod +from functools import partial +from multiprocessing import Pool +from typing import Dict, Iterator, List, Optional, Sequence, Tuple, Union + +import h5py +import numpy as np + +from . import _mat_v73 +from ._feature_chunking import DEFAULT_TARGET_CHUNK_BYTES + +__all__ = [ + "FeatureStore", + "HDF5FeatureStore", + "MatFeatureStore", + "detect_format", +] + +#: Root attribute marking a file as bdpy chunked feature storage. +FORMAT_ATTR = "bdpy_format" +#: Root attribute holding the schema version. +FORMAT_VERSION_ATTR = "bdpy_format_version" +#: Value of ``FORMAT_ATTR`` for feature files. +FORMAT_NAME = "features" +#: Highest schema version this module can read. +SUPPORTED_FORMAT_VERSION = 1 +#: Dataset holding the feature array, shape ``(n_samples, *feature_shape)``. +FEATURES_DATASET = "features" +#: Dataset holding the stimulus labels, shape ``(n_samples,)``. +LABELS_DATASET = "labels" +#: Extension of a chunked feature file. +HDF5_EXT = "h5" + +# A slice spec for the feature axes: what ``numpy.s_[...]`` produces. +FeatureSlice = Union[slice, int, Sequence[int], np.ndarray, Tuple, None] + + +def _load_array_with_key(key: str, path: str) -> np.ndarray: + # v5 .mat via scipy, v7.3 (HDF5) via h5py; avoids hdf5storage on the load + # path, which breaks under NumPy 2.0 (see bdpy/dataform/_mat_v73.py). + # NOTE: key comes first so that partial(..., 'feat') is a Pool.map callable. + return _mat_v73.loadmat_key(path, key) + + +def _determine_num_parallel(num_files: int) -> int: + # NOTE: optimal number of parallel processes is not clear. It could depend + # on several factors such as the number of files, the size of files, the + # number of cores, etc. For now, we use a simple heuristic based on the + # number of files. + num_parallel: int + if num_files < 16: + num_parallel = 1 + elif num_files < 64: + num_parallel = 16 + else: + num_parallel = 64 + return num_parallel + + +def _normalize_feature_slice(feature_slice: FeatureSlice) -> Tuple: + """Normalize a feature-axis slice spec to a tuple of per-axis indexers.""" + if feature_slice is None: + return () + if isinstance(feature_slice, tuple): + return feature_slice + return (feature_slice,) + + +def _is_fancy(indexer: object) -> bool: + """Whether an indexer is a fancy (list/array) index rather than a slice.""" + if isinstance(indexer, (slice, int, np.integer)): + return False + return isinstance(indexer, (list, tuple, np.ndarray, range)) + + +class FeatureStore(ABC): + """Interface to a collection of DNN features on disk. + + A store exposes layers and stimulus labels, and reads features for a set of + labels with an optional slice along the feature axes. Implementations differ + only in how much I/O a given request costs. + """ + + @property + @abstractmethod + def layers(self) -> List[str]: + """List of DNN layers held by this store.""" + + @property + @abstractmethod + def labels(self) -> List[str]: + """List of stimulus labels, in the store's own order.""" + + @abstractmethod + def shape(self, layer: str) -> Tuple[int, ...]: + """Full shape ``(n_samples, *feature_shape)`` of `layer`.""" + + @abstractmethod + def dtype(self, layer: str) -> np.dtype: + """Dtype of the features in `layer`.""" + + @abstractmethod + def read( + self, + layer: str, + labels: Optional[Sequence[str]] = None, + feature_slice: FeatureSlice = None, + ) -> np.ndarray: + """Read features from `layer`. + + Parameters + ---------- + layer : str + DNN layer. + labels : sequence of str, optional + Stimulus labels to read. ``None`` reads every label in store order. + Otherwise rows come back **in the order given**, and repeated labels + yield repeated rows. + feature_slice : slice, int, array-like or tuple, optional + Index applied to the feature axes (axes 1 and up), as produced by + ``numpy.s_[...]``. ``None`` reads the whole feature tensor. + + Returns + ------- + numpy.ndarray + Array of shape ``(n_labels, *sliced_feature_shape)``. + """ + + def chunk_extent(self, layer: str, axis: int) -> Optional[int]: + """On-disk chunk extent of `layer` along `axis`, if the store has one. + + Returns ``None`` when the backend has no chunking to align with, in + which case :meth:`iter_chunks` falls back to a byte budget. + """ + return None + + def iter_chunks( + self, + layer: str, + labels: Optional[Sequence[str]] = None, + feature_slice: FeatureSlice = None, + axis: int = 1, + size: Optional[int] = None, + ) -> Iterator[Tuple[slice, np.ndarray]]: + """Iterate over `layer` in slabs along `axis`. + + Yields ``(sl, block)`` pairs, where `sl` is the slice applied to `axis` + of the selected array and `block` is that slab. The slice is yielded so + that callers can place results back without tracking offsets themselves. + + Parameters + ---------- + layer : str + DNN layer. + labels : sequence of str, optional + Stimulus labels, as in :meth:`read`. + feature_slice : slice, int, array-like or tuple, optional + Index applied to the feature axes before iterating. + axis : int, optional + Axis of the *selected* array to iterate over. Axis 0 is the sample + axis; the default, axis 1, is the outermost feature axis, which is + the one partial feature reads are about. + size : int, optional + Number of elements per slab. Defaults to the on-disk chunk extent + along `axis` when the backend has one, so that iteration is + chunk-aligned and every element is read exactly once; otherwise a + size derived from the same byte budget is used. + + Yields + ------ + tuple of (slice, numpy.ndarray) + The slice applied to `axis`, and the corresponding slab. + """ + selected_shape = self._selected_shape(layer, labels, feature_slice) + if axis < 0: + axis += len(selected_shape) + if not 0 <= axis < len(selected_shape): + raise ValueError( + "axis {} is out of range for selected shape {}".format( + axis, selected_shape + ) + ) + + length = selected_shape[axis] + if size is None: + size = self._default_iter_size(layer, axis, selected_shape) + if size < 1: + raise ValueError("size must be positive, got {}".format(size)) + + base = _normalize_feature_slice(feature_slice) + for start in range(0, length, size): + sl = slice(start, min(start + size, length)) + if axis == 0: + block_labels = ( + list(self.labels) if labels is None else list(labels) + )[sl] + yield sl, self.read(layer, block_labels, feature_slice) + else: + yield sl, self.read( + layer, labels, _compose_slice(base, axis - 1, sl) + ) + + def _selected_shape( + self, + layer: str, + labels: Optional[Sequence[str]], + feature_slice: FeatureSlice, + ) -> Tuple[int, ...]: + """Shape the selection would have, without reading the data.""" + full = self.shape(layer) + n_samples = full[0] if labels is None else len(labels) + feature_shape = _sliced_shape(full[1:], feature_slice) + return (n_samples, *feature_shape) + + def _default_iter_size( + self, layer: str, axis: int, selected_shape: Tuple[int, ...] + ) -> int: + extent = self.chunk_extent(layer, axis) + if extent is not None: + return max(1, min(extent, selected_shape[axis])) + # No on-disk chunking to align with: fall back to the byte budget. + itemsize = np.dtype(self.dtype(layer)).itemsize + other = 1 + for i, n in enumerate(selected_shape): + if i != axis: + other *= int(n) + per_element = max(1, other * itemsize) + return max(1, min(selected_shape[axis], DEFAULT_TARGET_CHUNK_BYTES // per_element)) + + +def _compose_slice(base: Tuple, axis: int, sl: slice) -> Tuple: + """Return `base` with `sl` intersected into its entry for `axis`. + + `axis` indexes the *feature* axes (axis 0 here is array axis 1), and `sl` + is relative to the result of applying `base`. + """ + out = list(base) + [slice(None)] * max(0, axis + 1 - len(base)) + current = out[axis] + if isinstance(current, slice) and current == slice(None): + out[axis] = sl + elif isinstance(current, slice): + # Compose the two slices by indexing a range with both. + out[axis] = _compose_two_slices(current, sl) + else: + out[axis] = np.asarray(current)[sl] + return tuple(out) + + +def _compose_two_slices(outer: slice, inner: slice) -> slice: + """Compose ``x[outer][inner]`` into a single slice where possible.""" + # Only the simple (positive-step) case arises from iter_chunks, which + # always produces contiguous forward slices. + ostep = 1 if outer.step is None else outer.step + if ostep < 0: + # Fall back to an explicit index array for reversed slices. + raise ValueError("cannot compose a reversed slice; pass an explicit size") + ostart = 0 if outer.start is None else outer.start + start = ostart + (0 if inner.start is None else inner.start) * ostep + stop = ostart + (inner.stop if inner.stop is not None else 0) * ostep + if outer.stop is not None: + stop = min(stop, outer.stop) + return slice(start, stop, ostep) + + +def _sliced_shape( + feature_shape: Tuple[int, ...], feature_slice: FeatureSlice +) -> Tuple[int, ...]: + """Shape of ``numpy.empty(feature_shape)[feature_slice]``, cheaply.""" + indexers = _normalize_feature_slice(feature_slice) + if not indexers: + return feature_shape + # A broadcast dummy allocates nothing but lets NumPy do the index + # arithmetic, including negative steps, ellipses and newaxis. + dummy = np.broadcast_to(np.int8(0), feature_shape) + return tuple(int(n) for n in dummy[indexers].shape) + + +class MatFeatureStore(FeatureStore): + """Legacy per-stimulus ``.mat`` feature directory. + + Layout is ``//