From 60d82908422a145076f8db4cde2c2eb95e1cc825 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Thu, 9 Jul 2026 14:32:41 +0200 Subject: [PATCH] Bound memory use in h5grove structured content endpoint Any request to GET /api/datasets/{id}/content/{content_type} could read an entire HDF5 dataset into memory: h5grove's DatasetContent.data() with no selection executes dataset[()], selections are unbounded, data_stats reads everything, attributes() reads all attribute values and group metadata grows with child count. Cap in-memory reads at 1 MB: - data (format=bin/npy/csv) and stats now stream/accumulate in <= 1 MB slabs, so datasets of any size remain fully usable (h5web requests numeric data as format=bin) - data (format=json/tiff), attr and group meta reject oversized requests with a 400 suggesting the selection parameter, computed pre-read from shape, dtype itemsize and the parsed selection - variable-length dtypes cannot be pre-sized (h5py exposes no H5Dvlen_get_buf_size), so they are capped at ~1000 elements - invalid selection strings now return 400 instead of 500 --- lib/galaxy/datatypes/binary.py | 345 +++++++++++++++++- lib/galaxy/webapps/galaxy/api/datasets.py | 4 +- .../webapps/galaxy/services/datasets.py | 4 + test/integration/test_structured_dataset.py | 139 +++++-- test/unit/data/datatypes/test_h5.py | 275 ++++++++++++++ 5 files changed, 721 insertions(+), 46 deletions(-) create mode 100644 test/unit/data/datatypes/test_h5.py diff --git a/lib/galaxy/datatypes/binary.py b/lib/galaxy/datatypes/binary.py index f950d0fc84d..b3cb3cadde3 100644 --- a/lib/galaxy/datatypes/binary.py +++ b/lib/galaxy/datatypes/binary.py @@ -5,6 +5,7 @@ import gzip import io import json import logging +import math import os import re import shutil @@ -13,7 +14,10 @@ import subprocess import tarfile import tempfile import zipfile -from collections.abc import Iterable +from collections.abc import ( + Iterable, + Iterator, +) from json import dumps from typing import ( Any, @@ -33,9 +37,15 @@ from bx.seq.twobit import ( from h5grove.content import ( DatasetContent, get_content_from_file, + GroupContent, ResolvedEntityContent, ) from h5grove.encoders import encode +from h5grove.utils import ( + convert, + parse_slice, + QueryArgumentError, +) from galaxy import util from galaxy.datatypes import metadata @@ -78,6 +88,7 @@ from galaxy.datatypes.sniff import ( FilePrefix, ) from galaxy.datatypes.text import Html +from galaxy.exceptions import RequestParameterInvalidException from galaxy.util import ( compression_utils, nice_size, @@ -1249,6 +1260,292 @@ class BcfUncompressed(BaseBcf): return False +MAX_STRUCTURED_CONTENT_BYTES = 1000000 # 1 MB, matches DEFAULT_MAX_PEEK_SIZE in galaxy.datatypes.data +MAX_STRUCTURED_CONTENT_CHILDREN = MAX_STRUCTURED_CONTENT_BYTES // 128 # ~128 bytes/child metadata entry +VLEN_ELEMENT_ASSUMED_BYTES = 1024 + + +def _h5_normalize_selection(shape: tuple[int, ...], selection: Optional[str]) -> tuple[list[slice], tuple[int, ...]]: + """Normalize an h5grove ``selection`` string against a dataset ``shape``. + + Returns a per-axis list of read slices and the result shape. Integer indices + become width-1 slices whose axis is dropped from the result shape, matching + NumPy indexing. Unmentioned trailing axes are read in full. + + >>> _h5_normalize_selection((), None) + ([], ()) + >>> _h5_normalize_selection((10,), None) + ([slice(0, 10, 1)], (10,)) + >>> _h5_normalize_selection((10,), "5") + ([slice(5, 6, 1)], ()) + >>> _h5_normalize_selection((10,), "-3") + ([slice(7, 8, 1)], ()) + >>> _h5_normalize_selection((10,), "0:10:2") + ([slice(0, 10, 2)], (5,)) + >>> _h5_normalize_selection((10,), "3:") + ([slice(3, 10, 1)], (7,)) + >>> _h5_normalize_selection((4, 5), "1") + ([slice(1, 2, 1), slice(0, 5, 1)], (5,)) + """ + if selection is None: + return [slice(*slice(None).indices(dim)) for dim in shape], tuple(shape) + + try: + members = parse_slice(selection) + except (ValueError, TypeError) as e: + raise RequestParameterInvalidException(f"Invalid selection {selection!r}: {e}") + + if len(members) > len(shape): + raise RequestParameterInvalidException( + f"Selection {selection!r} has too many members for a {len(shape)}D dataset" + ) + + read_slices = [] + result_shape = [] + for axis, dim in enumerate(shape): + if axis >= len(members): + read_slices.append(slice(*slice(None).indices(dim))) + result_shape.append(dim) + continue + member = members[axis] + if isinstance(member, slice): + if member.step is not None and member.step <= 0: + raise RequestParameterInvalidException( + f"Selection {selection!r} has a non-positive step; only positive steps are supported" + ) + start, stop, step = member.indices(dim) + read_slices.append(slice(start, stop, step)) + result_shape.append(len(range(start, stop, step))) + else: + index = member + dim if member < 0 else member + if not 0 <= index < dim: + raise RequestParameterInvalidException( + f"Index {member} in selection {selection!r} is out of bounds for axis {axis} with size {dim}" + ) + read_slices.append(slice(index, index + 1, 1)) + return read_slices, tuple(result_shape) + + +def _h5_check_dataset_size(ds: h5py.Dataset, selection: Optional[str]) -> None: + if ds.shape is None: # null dataspace (h5py.Empty): holds no data + return + _, result_shape = _h5_normalize_selection(ds.shape, selection) + count = math.prod(result_shape) + if h5py.check_vlen_dtype(ds.dtype) is not None: + cap = MAX_STRUCTURED_CONTENT_BYTES // VLEN_ELEMENT_ASSUMED_BYTES + if count > cap: + raise RequestParameterInvalidException( + f"The selected data holds {count} variable-length elements, exceeding the limit of {cap}; " + "narrow the request with 'selection'." + ) + return + nbytes = count * ds.dtype.itemsize + if nbytes > MAX_STRUCTURED_CONTENT_BYTES: + raise RequestParameterInvalidException( + f"The selected data holds {nbytes} bytes, exceeding the limit of {MAX_STRUCTURED_CONTENT_BYTES} bytes; " + "narrow the request with 'selection'." + ) + + +def _h5_check_attributes_size(entity: h5py.HLObject) -> None: + total = 0 + for name in entity.attrs.keys(): + attr_id = entity.attrs.get_id(name) + total += math.prod(attr_id.shape or ()) * attr_id.get_type().get_size() + if total > MAX_STRUCTURED_CONTENT_BYTES: + raise RequestParameterInvalidException( + f"The attributes hold at least {total} bytes, exceeding the limit of " + f"{MAX_STRUCTURED_CONTENT_BYTES} bytes." + ) + + +def _h5_check_group_size(group: h5py.Group) -> None: + children = len(group) + if children > MAX_STRUCTURED_CONTENT_CHILDREN: + raise RequestParameterInvalidException( + f"The group has {children} children, exceeding the limit of {MAX_STRUCTURED_CONTENT_CHILDREN}; " + "browse subgroups individually." + ) + + +def _h5_iter_slabs(ds: h5py.Dataset, read_slices: list[slice]) -> Iterator[np.ndarray]: + itemsize = ds.dtype.itemsize + + def slabs(prefix: tuple[slice, ...], remaining: list[slice]) -> Iterator[np.ndarray]: + if not remaining: + yield ds[prefix] + return + first = remaining[0] + rest = remaining[1:] + # Iterate the first axis arithmetically, never materializing its indices + # (which could be billions). Positive step is guaranteed by + # _h5_normalize_selection, so batch_stop below never over-selects. + n = len(range(first.start, first.stop, first.step)) + row_nbytes = math.prod(len(range(s.start, s.stop, s.step)) for s in rest) * itemsize + if row_nbytes <= MAX_STRUCTURED_CONTENT_BYTES: + rows_per_batch = max(1, MAX_STRUCTURED_CONTENT_BYTES // max(row_nbytes, 1)) + for i in range(0, n, rows_per_batch): + batch_start = first.start + i * first.step + batch_n = min(rows_per_batch, n - i) + batch_stop = batch_start + batch_n * first.step + yield ds[prefix + (slice(batch_start, batch_stop, first.step),) + tuple(rest)] + else: + # A single row along this axis already exceeds the budget; fix it and + # recurse into the trailing axes so every read stays within the limit. + for i in range(n): + index = first.start + i * first.step + yield from slabs(prefix + (slice(index, index + 1, 1),), rest) + + if not read_slices: + yield ds[()] + else: + yield from slabs((), read_slices) + + +def _h5_npy_header_bytes(out_dtype: np.dtype, shape: tuple[int, ...]) -> bytes: + buffer = io.BytesIO() + np.lib.format.write_array_header_1_0( + buffer, + {"descr": np.lib.format.dtype_to_descr(out_dtype), "fortran_order": False, "shape": shape}, + ) + return buffer.getvalue() + + +def _h5_stream_data( + file_name: str, + path: str, + read_slices: list[slice], + dtype: str, + format: str, + out_dtype: np.dtype, + npy_shape: tuple[int, ...], + flatten: bool, +) -> Iterator[bytes]: + # This generator owns its own file handle: the caller's get_content_from_file + # context closes before the streaming response body is iterated. + with h5py.File(file_name, "r", locking=False) as f: + ds = f[path] + if format == "npy": + yield _h5_npy_header_bytes(out_dtype, npy_shape) + for slab in _h5_iter_slabs(ds, read_slices): + converted = convert(slab, dtype) + if format == "csv": + data = np.ravel(converted) if flatten else converted + with io.BytesIO() as buffer: + np.savetxt(buffer, data, delimiter=",") + yield buffer.getvalue() + else: + yield np.ascontiguousarray(converted).tobytes() + + +def _h5_prepare_streaming_data( + file_name: str, + path: str, + ds: h5py.Dataset, + dtype: str, + format: str, + flatten: bool, + selection: Optional[str], +) -> tuple[Iterator[bytes], dict[str, str]]: + read_slices, result_shape = _h5_normalize_selection(ds.shape, selection) + try: + out_dtype = convert(np.empty((0,), ds.dtype), dtype).dtype + except QueryArgumentError as e: + raise RequestParameterInvalidException(str(e)) + + is_numeric = np.issubdtype(out_dtype, np.number) or np.issubdtype(out_dtype, np.bool_) + if format in ("npy", "csv") and not is_numeric: + raise RequestParameterInvalidException(f"Unsupported format {format!r} for non-numeric data") + if format == "csv" and len(result_shape) == 0: + raise RequestParameterInvalidException("CSV format is not supported for scalar datasets") + if format == "csv" and not flatten and len(result_shape) > 2: + raise RequestParameterInvalidException( + "CSV format supports at most 2 dimensions; use 'selection' or 'flatten'." + ) + + count = math.prod(result_shape) + npy_shape = (count,) if flatten and result_shape != () else result_shape + generator = _h5_stream_data(file_name, path, read_slices, dtype, format, out_dtype, npy_shape, flatten) + if format == "bin": + headers = { + "Content-Type": "application/octet-stream", + "Content-Length": str(count * out_dtype.itemsize), + } + elif format == "npy": + header_nbytes = len(_h5_npy_header_bytes(out_dtype, npy_shape)) + headers = { + "Content-Type": "application/octet-stream", + "Content-Disposition": 'attachment; filename="data.npy"', + "Content-Length": str(header_nbytes + count * out_dtype.itemsize), + } + else: # csv + headers = { + "Content-Type": "text/csv", + "Content-Disposition": 'attachment; filename="data.csv"', + } + return generator, headers + + +def _h5_incremental_stats(ds: h5py.Dataset, read_slices: list[slice]) -> dict[str, Optional[float]]: + is_float = np.issubdtype(ds.dtype, np.floating) + cast = float if is_float else int + count = 0 + empty_stats: dict[str, Optional[float]] = { + "strict_positive_min": None, + "positive_min": None, + "min": None, + "max": None, + "mean": None, + "std": None, + } + if ds.shape is None: # null dataspace (h5py.Empty): no elements + return empty_stats + total = 0.0 + total_sq = 0.0 + minimum = None + maximum = None + positive_min = None + strict_positive_min = None + for slab in _h5_iter_slabs(ds, read_slices): + values = np.asarray(slab) + if is_float: + values = values[np.isfinite(values)] + if values.size == 0: + continue + as_float = values.astype(np.float64) + count += values.size + total += float(as_float.sum()) + total_sq += float(np.square(as_float).sum()) + slab_min = values.min() + slab_max = values.max() + minimum = slab_min if minimum is None else min(minimum, slab_min) + maximum = slab_max if maximum is None else max(maximum, slab_max) + positive = values[values >= 0] + if positive.size: + slab_positive_min = positive.min() + positive_min = slab_positive_min if positive_min is None else min(positive_min, slab_positive_min) + strict_positive = values[values > 0] + if strict_positive.size: + slab_strict_min = strict_positive.min() + strict_positive_min = ( + slab_strict_min if strict_positive_min is None else min(strict_positive_min, slab_strict_min) + ) + if count == 0: + return empty_stats + assert minimum is not None and maximum is not None + mean = total / count + variance = total_sq / count - mean * mean + std = math.sqrt(variance) if variance > 0 else 0.0 + return { + "strict_positive_min": cast(strict_positive_min) if strict_positive_min is not None else None, + "positive_min": cast(positive_min) if positive_min is not None else None, + "min": cast(minimum), + "max": cast(maximum), + "mean": cast(mean), + "std": cast(std), + } + + class H5(Binary): """ Class describing an HDF5 file @@ -1295,36 +1592,58 @@ class H5(Binary): def get_structured_content( self, - dataset, - content_type=None, - path="/", - dtype="origin", - format="json", - flatten=False, - selection=None, + dataset: DatasetProtocol, + content_type: Optional[str] = None, + path: str = "/", + dtype: str = "origin", + format: str = "json", + flatten: Union[bool, str] = False, + selection: Optional[str] = None, **kwargs, - ): + ) -> tuple[Union[bytes, str, Iterator[bytes]], dict[str, str]]: """ Implements h5grove protocol (https://silx-kit.github.io/h5grove/). This allows the h5web visualization tool (https://github.com/silx-kit/h5web) to be used directly with Galaxy datasets. """ - with get_content_from_file(dataset.get_file_name(), path, self._create_error) as content: + flatten = str(flatten).lower() != "false" + file_name = dataset.get_file_name() + with get_content_from_file(file_name, path, self._create_error, h5py_options={"locking": False}) as content: if content_type == "attr": assert isinstance(content, ResolvedEntityContent) + _h5_check_attributes_size(content._h5py_entity) resp = encode(content.attributes(), "json") elif content_type == "meta": + if isinstance(content, GroupContent): + _h5_check_group_size(content._h5py_entity) resp = encode(content.metadata(), "json") elif content_type == "stats": assert isinstance(content, DatasetContent) - resp = encode(content.data_stats(selection), "json") - else: # default 'data' + ds = content._h5py_entity + if h5py.check_vlen_dtype(ds.dtype) is not None: + # Variable-length dtypes cannot be slab-read; use h5grove's guarded in-memory path. + _h5_check_dataset_size(ds, selection) + resp = encode(content.data_stats(selection), "json") + elif ds.shape is None: # null dataspace: no elements, so empty stats + resp = encode(_h5_incremental_stats(ds, []), "json") + else: + read_slices, _ = _h5_normalize_selection(ds.shape, selection) + resp = encode(_h5_incremental_stats(ds, read_slices), "json") + elif content_type in ("data", None) and format in ("bin", "npy", "csv"): assert isinstance(content, DatasetContent) + ds = content._h5py_entity + if h5py.check_vlen_dtype(ds.dtype) is None and ds.shape is not None: + return _h5_prepare_streaming_data(file_name, path, ds, dtype, format, flatten, selection) + _h5_check_dataset_size(ds, selection) + resp = encode(content.data(selection, flatten, dtype), format) + else: # default 'data' with json/tiff, or variable-length dtype + assert isinstance(content, DatasetContent) + _h5_check_dataset_size(content._h5py_entity, selection) resp = encode(content.data(selection, flatten, dtype), format) return resp.content, resp.headers - def _create_error(self, status_code, message): + def _create_error(self, status_code: int, message: str) -> Exception: return Exception(status_code, message) diff --git a/lib/galaxy/webapps/galaxy/api/datasets.py b/lib/galaxy/webapps/galaxy/api/datasets.py index 35488d66eed..1075bb7bf9c 100644 --- a/lib/galaxy/webapps/galaxy/api/datasets.py +++ b/lib/galaxy/webapps/galaxy/api/datasets.py @@ -496,7 +496,9 @@ class FastAPIDatasets: content_type: DatasetContentType = DatasetContentType.data, ): content, headers = self.service.get_structured_content(trans, dataset_id, content_type, **request.query_params) - return Response(content=content, headers=headers) + if isinstance(content, (bytes, str)): + return Response(content=content, headers=headers) + return StreamingResponse(content, headers=headers) @router.delete( "/api/datasets", diff --git a/lib/galaxy/webapps/galaxy/services/datasets.py b/lib/galaxy/webapps/galaxy/services/datasets.py index d3f7f8ee8b3..f919880faba 100644 --- a/lib/galaxy/webapps/galaxy/services/datasets.py +++ b/lib/galaxy/webapps/galaxy/services/datasets.py @@ -824,6 +824,10 @@ class DatasetsService(ServiceBase, UsesVisualizationMixin): """ Retrieves contents of a dataset. It is left to the datatype to decide how to interpret the content types. + + Returns a ``(content, headers)`` tuple. ``content`` is usually a ``bytes`` + body, but datatypes that stream large content may return an iterator of + ``bytes`` chunks, which the controller serves as a ``StreamingResponse``. """ headers = {} content: Any = "" diff --git a/test/integration/test_structured_dataset.py b/test/integration/test_structured_dataset.py index 4f21261aaef..d00da4e855c 100644 --- a/test/integration/test_structured_dataset.py +++ b/test/integration/test_structured_dataset.py @@ -4,7 +4,13 @@ This file checks the ability to access datasets using the get_structured_content API and services. """ +import io import os +import tempfile +from urllib.parse import urlencode + +import h5py +import numpy as np from galaxy_test.base.populators import DatasetPopulator from galaxy_test.driver import integration_util @@ -12,61 +18,130 @@ from galaxy_test.driver import integration_util SCRIPT_DIR = os.path.normpath(os.path.dirname(__file__)) TEST_DATA_DIRECTORY = os.path.join(SCRIPT_DIR, os.pardir, os.pardir, "test-data") +BIG_ARRAY = np.arange(2_000_000, dtype=np.uint8) % 251 + + +def _build_structured_fixture(path): + with h5py.File(path, "w", libver="latest") as f: + f.create_dataset("big", data=BIG_ARRAY) + f["big"].attrs["big_attr"] = np.zeros(2_000_000, dtype=np.uint8) + f.create_dataset("small_vlen", data=["a", "bb", "ccc"]) + f.create_dataset("big_vlen", data=[str(i) for i in range(2000)]) + class TestStructuredDataset(integration_util.IntegrationTestCase): require_admin_user = True dataset_populator: DatasetPopulator test_history_id: str + _fixture_path = None + _upload_cache: dict = {} def setUp(self): super().setUp() self.dataset_populator = DatasetPopulator(self.galaxy_interactor) self.test_history_id = self.dataset_populator.new_history() + @classmethod + def _structured_fixture_path(cls): + if cls._fixture_path is None: + path = os.path.join(tempfile.mkdtemp(), "structured.h5") + _build_structured_fixture(path) + cls._fixture_path = path + return cls._fixture_path + + def _dataset_id(self, source_uri, file_type="h5"): + if source_uri not in type(self)._upload_cache: + dataset = self.dataset_populator.new_dataset( + self.test_history_id, source_uri, file_type=file_type, wait=True + ) + type(self)._upload_cache[source_uri] = dataset["dataset_id"] + return type(self)._upload_cache[source_uri] + + def _chopper_dataset_id(self): + return self._dataset_id(f"file://{TEST_DATA_DIRECTORY}/chopper.h5") + + def _big_dataset_id(self): + return self._dataset_id(f"file://{self._structured_fixture_path()}") + + def _content(self, dataset_id, content_type, **params): + query = f"?{urlencode(params)}" if params else "" + return self._get(f"datasets/{dataset_id}/content/{content_type}{query}") + def test_fail_on_nonbinary(self): - dataset = self.dataset_populator.new_dataset( - self.test_history_id, f"file://{TEST_DATA_DIRECTORY}/random-file", file_type="txt", wait=True - ) - dataset_id = dataset["dataset_id"] - response = self._get(f"datasets/{dataset_id}/content/meta") + dataset_id = self._dataset_id(f"file://{TEST_DATA_DIRECTORY}/random-file", file_type="txt") + response = self._content(dataset_id, "meta") self._assert_status_code_is(response, 500) def test_api_meta(self): - dataset = self.dataset_populator.new_dataset( - self.test_history_id, f"file://{TEST_DATA_DIRECTORY}/chopper.h5", file_type="h5", wait=True - ) - dataset_id = dataset["dataset_id"] - response = self._get(f"datasets/{dataset_id}/content/meta") + response = self._content(self._chopper_dataset_id(), "meta") self._assert_status_code_is(response, 200) - hvals = response.json() - self._assert_has_keys(hvals, "attributes", "name", "kind") + self._assert_has_keys(response.json(), "attributes", "name", "kind") def test_api_attr(self): - dataset = self.dataset_populator.new_dataset( - self.test_history_id, f"file://{TEST_DATA_DIRECTORY}/chopper.h5", file_type="h5", wait=True - ) - dataset_id = dataset["dataset_id"] - response = self._get(f"datasets/{dataset_id}/content/attr") + response = self._content(self._chopper_dataset_id(), "attr") self._assert_status_code_is(response, 200) - hvals = response.json() - self._assert_has_keys(hvals, "HDF5_Version", "NeXus_version", "default", "file_name", "file_time") + self._assert_has_keys(response.json(), "HDF5_Version", "NeXus_version", "default", "file_name", "file_time") def test_api_stats(self): - dataset = self.dataset_populator.new_dataset( - self.test_history_id, f"file://{TEST_DATA_DIRECTORY}/chopper.h5", file_type="h5", wait=True - ) - dataset_id = dataset["dataset_id"] - response = self._get(f"datasets/{dataset_id}/content/stats?path=%2Fentry%2Fdata%2Fdata") + response = self._content(self._chopper_dataset_id(), "stats", path="/entry/data/data") self._assert_status_code_is(response, 200) - hvals = response.json() - self._assert_has_keys(hvals, "strict_positive_min", "positive_min", "min", "max", "mean", "std") + self._assert_has_keys(response.json(), "strict_positive_min", "positive_min", "min", "max", "mean", "std") def test_api_data(self): - dataset = self.dataset_populator.new_dataset( - self.test_history_id, f"file://{TEST_DATA_DIRECTORY}/chopper.h5", file_type="h5", wait=True - ) - dataset_id = dataset["dataset_id"] - response = self._get(f"datasets/{dataset_id}/content/data?path=%2Fentry%2Fdata%2Fdata") + response = self._content(self._chopper_dataset_id(), "data", path="/entry/data/data") + self._assert_status_code_is(response, 200) + assert len(response.json()) == 148 + + def test_data_json_over_limit_rejected(self): + response = self._content(self._big_dataset_id(), "data", path="/big") + self._assert_status_code_is(response, 400) + assert "selection" in response.text.lower() + + def test_data_json_with_selection(self): + response = self._content(self._big_dataset_id(), "data", path="/big", selection="0:100") + self._assert_status_code_is(response, 200) + assert response.json() == BIG_ARRAY[0:100].tolist() + + def test_data_bin_streams_over_limit(self): + response = self._content(self._big_dataset_id(), "data", path="/big", format="bin") + self._assert_status_code_is(response, 200) + assert len(response.content) == 2_000_000 + assert int(response.headers["content-length"]) == 2_000_000 + assert response.content == BIG_ARRAY.tobytes() + + def test_data_npy_round_trips(self): + response = self._content(self._big_dataset_id(), "data", path="/big", format="npy") + self._assert_status_code_is(response, 200) + loaded = np.load(io.BytesIO(response.content)) + assert loaded.shape == (2_000_000,) + assert np.array_equal(loaded, BIG_ARRAY) + + def test_stats_incremental_over_limit(self): + response = self._content(self._big_dataset_id(), "stats", path="/big") self._assert_status_code_is(response, 200) hvals = response.json() - assert len(hvals) == 148 + assert hvals["min"] == int(BIG_ARRAY.min()) + assert hvals["max"] == int(BIG_ARRAY.max()) + assert hvals["mean"] == int(BIG_ARRAY.mean()) + assert hvals["std"] == int(BIG_ARRAY.std()) + + def test_data_invalid_selection_rejected(self): + response = self._content(self._big_dataset_id(), "data", path="/big", selection="foo") + self._assert_status_code_is(response, 400) + + def test_data_nonpositive_step_selection_rejected(self): + response = self._content(self._big_dataset_id(), "data", path="/big", selection="0:10:0") + self._assert_status_code_is(response, 400) + + def test_attr_over_limit_rejected(self): + response = self._content(self._big_dataset_id(), "attr", path="/big") + self._assert_status_code_is(response, 400) + + def test_data_vlen_under_cap(self): + response = self._content(self._big_dataset_id(), "data", path="/small_vlen") + self._assert_status_code_is(response, 200) + assert len(response.json()) == 3 + + def test_data_vlen_over_cap_rejected(self): + response = self._content(self._big_dataset_id(), "data", path="/big_vlen") + self._assert_status_code_is(response, 400) diff --git a/test/unit/data/datatypes/test_h5.py b/test/unit/data/datatypes/test_h5.py new file mode 100644 index 00000000000..b91e5ab1ee0 --- /dev/null +++ b/test/unit/data/datatypes/test_h5.py @@ -0,0 +1,275 @@ +import io +import json + +import h5py +import numpy as np +import pytest +from h5grove.utils import get_array_stats + +from galaxy.datatypes.binary import ( + _h5_incremental_stats, + _h5_iter_slabs, + _h5_normalize_selection, + H5, + MAX_STRUCTURED_CONTENT_BYTES, +) +from galaxy.exceptions import RequestParameterInvalidException + +EMPTY_STATS = { + "strict_positive_min": None, + "positive_min": None, + "min": None, + "max": None, + "mean": None, + "std": None, +} + + +class FakeDataset: + def __init__(self, file_name): + self.file_name_ = file_name + + def get_file_name(self, sync_cache=True): + return self.file_name_ + + +def make_h5(path, builder): + with h5py.File(path, "w", libver="latest") as f: + builder(f) + return FakeDataset(str(path)) + + +def single_dataset_h5(path, data, name="big"): + return make_h5(path, lambda f: f.create_dataset(name, data=data)) + + +def stream_slabs(dataset, selection=None, name="big"): + with h5py.File(dataset.get_file_name(), "r") as f: + read_slices, result_shape = _h5_normalize_selection(f[name].shape, selection) + chunks = list(_h5_iter_slabs(f[name], read_slices)) + joined = b"".join(np.ascontiguousarray(chunk).tobytes() for chunk in chunks) + max_nbytes = max((chunk.nbytes for chunk in chunks), default=0) + return joined, result_shape, max_nbytes + + +def compute_stats(dataset, selection=None, name="big"): + with h5py.File(dataset.get_file_name(), "r") as f: + read_slices, _ = _h5_normalize_selection(f[name].shape, selection) + return _h5_incremental_stats(f[name], read_slices) + + +def structured_content(dataset, content_type, **kwargs): + return H5().get_structured_content(dataset, content_type=content_type, **kwargs) + + +def assert_stats_equal(actual, expected): + for key, expected_value in expected.items(): + if expected_value is None: + assert actual[key] is None + else: + assert actual[key] == pytest.approx(expected_value, rel=1e-6, abs=1e-6) + + +def test_normalize_selection_multiaxis_mixed(): + read_slices, result_shape = _h5_normalize_selection((4, 8), "1:4,2:8:2") + assert read_slices == [slice(1, 4, 1), slice(2, 8, 2)] + assert result_shape == (3, 3) + + +def test_normalize_selection_trailing_axis_full(): + read_slices, result_shape = _h5_normalize_selection((4, 8), "2") + assert read_slices == [slice(2, 3, 1), slice(0, 8, 1)] + assert result_shape == (8,) + + +def test_normalize_selection_invalid_string_raises(): + with pytest.raises(RequestParameterInvalidException): + _h5_normalize_selection((10,), "foo") + + +def test_normalize_selection_too_many_members_raises(): + with pytest.raises(RequestParameterInvalidException): + _h5_normalize_selection((10,), "1,2") + + +def test_normalize_selection_out_of_bounds_index_raises(): + with pytest.raises(RequestParameterInvalidException): + _h5_normalize_selection((10,), "10") + + +@pytest.mark.parametrize("selection", ["0:10:0", "0:10:-1", "::-2"]) +def test_normalize_selection_non_positive_step_rejected(selection): + with pytest.raises(RequestParameterInvalidException): + _h5_normalize_selection((10,), selection) + + +def test_slabs_zero_length_axis(tmp_path): + dataset = single_dataset_h5(tmp_path / "z.h5", np.zeros((10, 0), dtype=np.uint8)) + joined, _, max_nbytes = stream_slabs(dataset) + assert joined == b"" + assert max_nbytes <= MAX_STRUCTURED_CONTENT_BYTES + + +def test_slabs_large_first_axis_with_step(tmp_path): + data = np.arange(3_000_000, dtype=np.uint8) + dataset = single_dataset_h5(tmp_path / "step.h5", data) + joined, _, max_nbytes = stream_slabs(dataset, "0:3000000:2") + assert max_nbytes <= MAX_STRUCTURED_CONTENT_BYTES + assert joined == data[0:3000000:2].tobytes() + + +def test_slabs_1d_bounded_and_lossless(tmp_path): + data = np.arange(2_500_000, dtype=np.uint8) + dataset = single_dataset_h5(tmp_path / "d.h5", data) + joined, _, max_nbytes = stream_slabs(dataset) + assert max_nbytes <= MAX_STRUCTURED_CONTENT_BYTES + assert joined == data.tobytes() + + +def test_slabs_wide_row_forces_trailing_recursion(tmp_path): + data = np.arange(4 * 2_000_000, dtype=np.uint8).reshape(4, 2_000_000) + dataset = single_dataset_h5(tmp_path / "wide.h5", data) + joined, _, max_nbytes = stream_slabs(dataset) + assert max_nbytes <= MAX_STRUCTURED_CONTENT_BYTES + assert joined == data.tobytes() + + +def test_slabs_step_and_int_index(tmp_path): + data = np.arange(10 * 8, dtype=np.int32).reshape(10, 8) + dataset = single_dataset_h5(tmp_path / "s.h5", data) + joined, result_shape, _ = stream_slabs(dataset, "0:10:2,3") + assert result_shape == (5,) + assert joined == data[0:10:2, 3].tobytes() + + +@pytest.mark.parametrize("flatten", [False, True]) +def test_npy_streaming_roundtrip(tmp_path, flatten): + data = np.arange(300 * 400, dtype=np.float32).reshape(300, 400) + dataset = single_dataset_h5(tmp_path / "npy.h5", data) + content, headers = structured_content( + dataset, "data", path="/big", format="npy", flatten="true" if flatten else "false" + ) + body = b"".join(content) + assert int(headers["Content-Length"]) == len(body) + loaded = np.load(io.BytesIO(body)) + expected = data.ravel() if flatten else data + assert loaded.shape == expected.shape + assert np.array_equal(loaded, expected) + + +def test_bin_streaming_exact_length(tmp_path): + data = np.arange(2_000_000, dtype=np.uint8) + dataset = single_dataset_h5(tmp_path / "bin.h5", data) + content, headers = structured_content(dataset, "data", path="/big", format="bin") + body = b"".join(content) + assert len(body) == 2_000_000 + assert int(headers["Content-Length"]) == 2_000_000 + assert body == data.tobytes() + + +def test_csv_streaming_roundtrip(tmp_path): + data = np.arange(300 * 500, dtype=np.float64).reshape(300, 500) + dataset = single_dataset_h5(tmp_path / "csv.h5", data) + content, headers = structured_content(dataset, "data", path="/big", format="csv") + body = b"".join(content) + assert headers["Content-Type"] == "text/csv" + loaded = np.loadtxt(io.BytesIO(body), delimiter=",") + assert loaded.shape == (300, 500) + assert np.allclose(loaded, data) + + +def test_csv_rejects_more_than_two_dimensions(tmp_path): + dataset = single_dataset_h5(tmp_path / "c3.h5", np.zeros((4, 4, 4), dtype=np.float64)) + with pytest.raises(RequestParameterInvalidException): + structured_content(dataset, "data", path="/big", format="csv") + + +def test_csv_rejects_non_numeric(tmp_path): + dataset = single_dataset_h5(tmp_path / "cs.h5", np.array([b"abc", b"def"], dtype="S3")) + with pytest.raises(RequestParameterInvalidException): + structured_content(dataset, "data", path="/big", format="csv") + + +def test_incremental_stats_float_with_nan_inf(tmp_path): + rng = np.random.default_rng(0) + data = rng.standard_normal(500_000).astype(np.float64) + data[10] = np.nan + data[20] = np.inf + data[30] = -np.inf + dataset = single_dataset_h5(tmp_path / "f.h5", data) + assert_stats_equal(compute_stats(dataset), get_array_stats(data[np.isfinite(data)])) + + +def test_incremental_stats_int(tmp_path): + data = np.arange(-100, 100, dtype=np.int64) + dataset = single_dataset_h5(tmp_path / "i.h5", data) + assert_stats_equal(compute_stats(dataset), get_array_stats(data)) + + +def test_incremental_stats_all_nan(tmp_path): + dataset = single_dataset_h5(tmp_path / "n.h5", np.full(1000, np.nan, dtype=np.float64)) + assert compute_stats(dataset) == EMPTY_STATS + + +def test_over_limit_json_data_rejected(tmp_path): + dataset = single_dataset_h5(tmp_path / "big.h5", np.zeros(2_000_000, dtype=np.uint8)) + with pytest.raises(RequestParameterInvalidException): + structured_content(dataset, "data", path="/big", format="json") + + +def test_over_limit_json_data_allowed_with_selection(tmp_path): + data = np.arange(2_000_000, dtype=np.uint8) % 251 + dataset = single_dataset_h5(tmp_path / "big.h5", data) + content, _ = structured_content(dataset, "data", path="/big", format="json", selection="0:100") + assert json.loads(content) == data[0:100].tolist() + + +def test_invalid_selection_rejected(tmp_path): + dataset = single_dataset_h5(tmp_path / "s.h5", np.zeros(100, dtype=np.uint8)) + with pytest.raises(RequestParameterInvalidException): + structured_content(dataset, "data", path="/big", format="json", selection="foo") + + +def test_attribute_guard(tmp_path): + def builder(f): + ds = f.create_dataset("big", data=np.zeros(10, dtype=np.uint8)) + ds.attrs["huge"] = np.zeros(200_000, dtype=np.float64) + + dataset = make_h5(tmp_path / "a.h5", builder) + with pytest.raises(RequestParameterInvalidException): + structured_content(dataset, "attr", path="/big") + + +def test_group_guard(tmp_path, monkeypatch): + monkeypatch.setattr("galaxy.datatypes.binary.MAX_STRUCTURED_CONTENT_CHILDREN", 3) + + def builder(f): + group = f.create_group("g") + for i in range(5): + group.create_dataset(f"d{i}", data=np.zeros(1, dtype=np.uint8)) + + dataset = make_h5(tmp_path / "g.h5", builder) + with pytest.raises(RequestParameterInvalidException): + structured_content(dataset, "meta", path="/g") + + +def test_vlen_over_limit_rejected(tmp_path): + def builder(f): + ds = f.create_dataset("v", (2000,), dtype=h5py.string_dtype()) + ds[...] = np.array(["x"] * 2000, dtype=object) + + dataset = make_h5(tmp_path / "v.h5", builder) + with pytest.raises(RequestParameterInvalidException): + structured_content(dataset, "data", path="/v", format="json") + + +def test_null_dataspace_data_json(tmp_path): + dataset = make_h5(tmp_path / "e.h5", lambda f: f.create_dataset("empty", dtype="f8")) + content, _ = structured_content(dataset, "data", path="/empty", format="json") + assert json.loads(content) is None + + +def test_null_dataspace_stats(tmp_path): + dataset = make_h5(tmp_path / "e.h5", lambda f: f.create_dataset("empty", dtype="f8")) + content, _ = structured_content(dataset, "stats", path="/empty") + assert json.loads(content) == EMPTY_STATS