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
This commit is contained in:
mvdbeek
2026-07-09 19:17:31 +02:00
parent 867820ddb5
commit 60d8290842
5 changed files with 721 additions and 46 deletions
+332 -13
View File
@@ -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)
+3 -1
View File
@@ -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",
@@ -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 = ""
+107 -32
View File
@@ -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)
+275
View File
@@ -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