From ae8577be2950f167042e6195cc5ed8b554075da6 Mon Sep 17 00:00:00 2001 From: Francesc Alted Date: Thu, 3 Sep 2026 13:37:57 +0200 Subject: [PATCH 01/10] Unify lazy caching for remote arrays --- doc/guides/remote_arrays.md | 34 +++++++-- examples/s3-cat2-access.py | 108 +++++++++++++++++++++++++++ src/blosc2/c2array.py | 5 ++ src/blosc2/schunk.py | 63 ++++++++++++---- tests/ndarray/test_c2array_blocks.py | 82 ++++++++++++++++++++ 5 files changed, 274 insertions(+), 18 deletions(-) create mode 100644 examples/s3-cat2-access.py diff --git a/doc/guides/remote_arrays.md b/doc/guides/remote_arrays.md index 2d4a03a3f..d0e43b7b6 100644 --- a/doc/guides/remote_arrays.md +++ b/doc/guides/remote_arrays.md @@ -7,7 +7,7 @@ A Blosc2 array that lives on a server does not have to be downloaded to be used. | Where the array lives | How to open it | |---|---| | Any URL fsspec reaches — `s3://`, `gs://`, `https://`, `zip://`… | `blosc2.open(url, lazy=True)` | -| A [Caterva2](https://ironarray.io/caterva2) server | `blosc2.C2Array(path, urlbase=...)` | +| A [Caterva2](https://ironarray.io/caterva2) server | `blosc2.open(blosc2.URLPath(path, urlbase=...), lazy=True)` | | Anything else | A `read_range()` of your own — see [Your own transport](#your-own-transport) | ```python @@ -16,9 +16,11 @@ import blosc2 # An object store, a web server, a zip on either of them a = blosc2.open("s3://bucket/big.b2nd", lazy=True) -# A Caterva2 server -b = blosc2.C2Array( - "@public/examples/lung-jpeg2000_10x.b2nd", urlbase="https://cat2.cloud/demo" +# A Caterva2 server; add lazy=True for an automatic Proxy cache +b = blosc2.open( + blosc2.URLPath( + "@public/examples/lung-jpeg2000_10x.b2nd", urlbase="https://cat2.cloud/demo" + ) ) a.shape, a.dtype # metadata only; nothing was downloaded @@ -46,7 +48,9 @@ p[10:12, 500:600] # fetched from the server, and written to lung-cache.b2nd That file is an ordinary Blosc2 array holding only the pieces you touched — a few hundred bytes for a freshly opened proxy over a 64 MB dataset, growing as you read. It is a normal `.b2nd`: copy it, ship it, open it with {func}`blosc2.open`. With `mode="a"` a later run picks up where the last one left off. -{func}`blosc2.open` builds the proxy for you and offers the same choice under another name — `cache_storage=` a directory for a cache on disk, nothing for one in memory: +{func}`blosc2.open` builds the proxy for either kind of remote source and offers +the same choice under another name — `cache_storage=` for a cache on disk, +nothing for one in memory: ```python url = "s3://bucket/big.b2nd" @@ -60,6 +64,25 @@ a = blosc2.open(url, lazy=True, cache_storage="./b2cache") a[100:110, :50] # no request ``` +The same interface works for Caterva2: + +```python +url = blosc2.URLPath("@personal/run.b2nd") + +with blosc2.c2context( + urlbase="https://cat2.cloud/demo", + username="me@example.com", + password="secret", +): + a = blosc2.open(url, lazy=True, cache_storage="./b2cache") + a[100:110, :50] +``` + +For authenticated Caterva2 datasets, `cache_storage` must be private to the +current user. Applications serving multiple users must use a separate cache +directory for each user; sharing one between users is not supported. Reopen a +private cache inside an equivalent authenticated {func}`c2context`. + ## Only what a slice touches A chunk is the unit a container is compressed in, and it can be several megabytes. Fetching a whole one to read a corner of it is most of the cost of a remote read, so Blosc2 fetches **blocks** — the smaller pieces a chunk is built from — whenever a slice lands in a small part of a large chunk. @@ -223,5 +246,6 @@ Four things to get right: - {doc}`Tutorial 6 <../tutorials/06.remote_proxy>` — the same ground at a slower pace, with output. - `examples/ndarray/rw-fsspec.py` — every way of reading and writing an fsspec URL, runnable. +- `examples/s3-cat2-access.py` — the same dataset and cache API through HTTPS/fsspec and Caterva2, with timings. - `examples/c2array-traffic.py` — what a remote slice costs in bytes, and what blocks and the cache save, runnable. - {ref}`C2Array`, {ref}`FsspecNDSource`, {ref}`ByteRangeNDSource`, {ref}`Proxy`, {ref}`Traffic` — the reference pages. diff --git a/examples/s3-cat2-access.py b/examples/s3-cat2-access.py new file mode 100644 index 000000000..c39733b46 --- /dev/null +++ b/examples/s3-cat2-access.py @@ -0,0 +1,108 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +"""Compare lazy access to the same array through fsspec and Caterva2. + +The HTTPS path needs the fsspec HTTP dependencies. Install them with: + + pip install "blosc2[fsspec]" aiohttp + +By default, caches are kept under ``./s3-cat2-cache``. Run the example again to +see the first data access served by the cache left by the previous process. +""" + +import argparse +from pathlib import Path +from time import perf_counter + +import numpy as np + +import blosc2 + +CATERVA2_URL = blosc2.URLPath( + "@public/examples/cube-1k-1k-1k.b2nd", + urlbase="https://cat2.cloud/demo", +) +# The same contents are published in this bucket with a ``-2`` suffix. +FSSPEC_URL = "https://blosc2.s3.us-west-001.backblazeb2.com/cube-1k-1k-1k-2.b2nd" +SLICE = np.s_[100:110, 200:300, 400:500] + + +def traffic_text(traffic: blosc2.Traffic | None) -> str: + if traffic is None: + return "traffic unavailable" + return f"{traffic.requests} requests, {traffic.nbytes / 2**20:.3f} MiB" + + +def size_text(size: int) -> str: + return f"{size / 2**20:.3f} MiB" + + +def benchmark(label: str, urlpath, cache_storage: Path) -> np.ndarray: + cache_existed = cache_storage.is_dir() and any(cache_storage.glob("*.b2nd")) + + start = perf_counter() + array = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + open_time = perf_counter() - start + + metadata = (array.shape, array.dtype, array.chunks, array.blocks) + cache_path = Path(array.urlpath).resolve() + + array.traffic.reset() + start = perf_counter() + data = array[SLICE] + first_read_time = perf_counter() - start + first_traffic = traffic_text(array.traffic) + cache_size = cache_path.stat().st_size + + # Open a fresh remote handle over the same on-disk cache. This demonstrates + # that cached data survives the Proxy object, not merely one array access. + del array + start = perf_counter() + reopened = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + reopen_time = perf_counter() - start + + reopened.traffic.reset() + start = perf_counter() + cached = reopened[SLICE] + cached_read_time = perf_counter() - start + cached_traffic = traffic_text(reopened.traffic) + np.testing.assert_array_equal(cached, data) + + print(f"\n{label}") + print(f" metadata: shape={metadata[0]}, dtype={metadata[1]}") + print(f" chunks={metadata[2]}, blocks={metadata[3]}") + print(f" persistent cache: {cache_path} ({'existing' if cache_existed else 'new'})") + print(f" open and remote metadata setup: {open_time:.6f} s") + print(f" first data slice this run: {first_read_time:.6f} s ({first_traffic})") + print(f" cache size after slice: {size_text(cache_size)}") + print(f" reopen persistent cache: {reopen_time:.6f} s") + print(f" same slice after reopen: {cached_read_time:.6f} s ({cached_traffic})") + return data + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--cache-dir", + type=Path, + default=Path("s3-cat2-cache"), + help="persistent cache root (default: ./s3-cat2-cache)", + ) + args = parser.parse_args() + root = args.cache_dir + + print(f"Persistent cache root: {root.resolve()}") + print("Run this command again to reuse these cache files.") + cat2_data = benchmark("Caterva2", CATERVA2_URL, root / "caterva2") + fsspec_data = benchmark("fsspec over HTTPS", FSSPEC_URL, root / "fsspec") + np.testing.assert_array_equal(cat2_data, fsspec_data) + print("\nBoth services returned identical data.") + + +if __name__ == "__main__": + main() diff --git a/src/blosc2/c2array.py b/src/blosc2/c2array.py index 9db9583eb..a2745a155 100644 --- a/src/blosc2/c2array.py +++ b/src/blosc2/c2array.py @@ -1670,6 +1670,11 @@ def __init__(self, path: str, /, urlbase: str | None = None, auth_token: str | N Create an instance of a remote data file (aka :ref:`C2Array `) urlpath. This is meant to be used in the :func:`blosc2.open` function. + Passing this object to :func:`blosc2.open` returns a :ref:`C2Array`. With + ``lazy=True`` it instead returns a :ref:`Proxy`, using an in-memory cache + by default or a persistent cache when ``cache_storage`` is provided. + Authenticated users sharing a machine must use separate cache directories. + The parameters are the same as for the :meth:`C2Array.__init__`. """ diff --git a/src/blosc2/schunk.py b/src/blosc2/schunk.py index 1db3449ca..402456fae 100644 --- a/src/blosc2/schunk.py +++ b/src/blosc2/schunk.py @@ -1987,11 +1987,17 @@ def _lazy_fsspec_proxy( # None leaves the default where it belongs, on the source itself kwargs = {} if max_concurrency is None else {"max_concurrency": max_concurrency} src = blosc2.FsspecNDSource(urlpath, **kwargs) + return _lazy_remote_proxy(src, urlpath, cache_storage) + + +def _lazy_remote_proxy(src, identity: str, cache_storage: str | pathlib.Path | None): + """Wrap a remote source in a memory or persistent cache.""" if cache_storage is None: return blosc2.Proxy(src) - path = fsspec_cache_path(urlpath, cache_storage, ".b2nd") - if os.path.exists(path) and _cache_stamp(path) != src.stamp: + path = fsspec_cache_path(identity, cache_storage, ".b2nd") + stamp = getattr(src, "stamp", None) + if os.path.exists(path) and _cache_stamp(path) != stamp: # The remote frame was replaced, which makes every cached chunk -- and # every offset they were fetched by -- meaningless blosc2.remove_urlpath(path) @@ -2000,6 +2006,34 @@ def _lazy_fsspec_proxy( return blosc2.Proxy(src, urlpath=path, mode="a") +def _open_c2_urlpath(urlpath: blosc2.URLPath, mode: str, offset: int, kwargs: dict): + """Open a Caterva2 array directly, or through the same lazy cache API as fsspec.""" + if mode != "r": + raise NotImplementedError(f"Caterva2 arrays can only be opened with mode='r', not {mode!r}") + if offset != 0: + raise NotImplementedError("offset is not supported for Caterva2 arrays") + + cache_storage = kwargs.pop("cache_storage", None) + max_concurrency = kwargs.pop("max_concurrency", None) + lazy = kwargs.pop("lazy", False) + requested = [key for key, value in kwargs.items() if value is not None] + if requested: + raise NotImplementedError(f"{', '.join(requested)} is not supported for Caterva2 arrays") + + if not lazy: + if cache_storage is not None: + raise NotImplementedError("cache_storage for a Caterva2 array requires lazy=True") + if max_concurrency is not None: + raise NotImplementedError("max_concurrency is only supported with lazy=True") + return blosc2.C2Array(urlpath.path, urlbase=urlpath.urlbase, auth_token=urlpath.auth_token) + + src = blosc2.C2Array(urlpath.path, urlbase=urlpath.urlbase, auth_token=urlpath.auth_token) + if max_concurrency is not None: + src.max_concurrency = max_concurrency + identity = f"caterva2:{blosc2.c2array._server_url(src.urlbase, src.path)}" + return _lazy_remote_proxy(src, identity, cache_storage) + + def _cache_stamp(path: str): """The remote stamp a cached proxy container was built against, if any. @@ -2098,8 +2132,9 @@ def open( Open modes also define the allowed persistence side effects: - - ``'r'`` never writes to the persistent object or any sidecar/cache file. - Query acceleration and other execution caches remain process-local only. + - ``'r'`` never writes to the persistent object. It writes a local cache + only when ``cache_storage`` explicitly requests one; query acceleration + and other implicit execution caches remain process-local only. - ``'a'`` and ``'w'`` may persist explicit user-visible changes such as data, metadata, and index maintenance, but execution caches and query memoization still remain process-local only. @@ -2108,7 +2143,8 @@ def open( (e.g. in a file containing several such objects). kwargs: dict, optional lazy: bool, optional - Only for fsspec URLs: return a :ref:`Proxy` that leaves the container + For fsspec URLs and Caterva2 :ref:`URLPath` objects, return a + :ref:`Proxy` that leaves the container where it is and reads what a slice touches, in range requests, instead of transferring the whole thing. Contiguous frames holding an :ref:`NDArray` only. A slice landing in a small part of a large chunk @@ -2125,7 +2161,8 @@ def open( hide, where the pool costs about 10 microseconds per chunk and saves nothing. cache_storage: str | pathlib.Path, optional - Only for fsspec URLs: a directory holding this container's local + For fsspec URLs and lazy Caterva2 :ref:`URLPath` objects, a directory + holding this container's local copy — the whole thing, or just the chunks and blocks ``lazy`` has fetched so far. Either way a later run starts from what is already there, and the copy is discarded when the remote no longer matches @@ -2157,7 +2194,8 @@ def open( Returns ------- - out: :ref:`SChunk`, :ref:`NDArray`, :ref:`C2Array`, :ref:`DictStore`, :ref:`EmbedStore`, or :ref:`TreeStore` + out: :ref:`SChunk`, :ref:`NDArray`, :ref:`C2Array`, :ref:`Proxy`, + :ref:`DictStore`, :ref:`EmbedStore`, or :ref:`TreeStore` The object found in the path. Notes @@ -2169,7 +2207,10 @@ def open( :class:`LazyArray`, exiting the context is currently a no-op. * If :paramref:`urlpath` is a :ref:`URLPath` instance, :paramref:`mode` - must be 'r', :paramref:`offset` must be 0, and kwargs cannot be passed. + must be 'r' and :paramref:`offset` must be 0. Without ``lazy=True`` it + returns a :ref:`C2Array`; with ``lazy=True`` it returns a :ref:`Proxy`, + optionally persisted under ``cache_storage``. Authenticated users sharing + a machine must use separate cache directories. * fsspec URLs need the ``fsspec`` extra (``pip install "blosc2[fsspec]"``) and the driver for the protocol (``s3fs``, ``gcsfs``...), which fsspec asks for @@ -2234,11 +2275,7 @@ def open( True """ if isinstance(urlpath, blosc2.URLPath): - if mode != "r" or offset != 0 or kwargs != {}: - raise NotImplementedError( - "Cannot open a C2Array with mode != 'r', or offset != 0 or some kwargs" - ) - return blosc2.C2Array(urlpath.path, urlbase=urlpath.urlbase, auth_token=urlpath.auth_token) + return _open_c2_urlpath(urlpath, mode, offset, kwargs) if isinstance(urlpath, pathlib.PurePath): urlpath = str(urlpath) diff --git a/tests/ndarray/test_c2array_blocks.py b/tests/ndarray/test_c2array_blocks.py index 4ca8deb40..a5fce661d 100644 --- a/tests/ndarray/test_c2array_blocks.py +++ b/tests/ndarray/test_c2array_blocks.py @@ -324,6 +324,88 @@ def _bytes(srv, endpoint): return sum(n for kind, _, n in srv.log if kind == endpoint) +def test_open_urlpath_lazy_memory_cache(server, any_chunk_wants_blocks): + data = _incompressible((200, 200)) + array, srv = server(data, chunks=(100, 200), blocks=(10, 20)) + urlpath = blosc2.URLPath(array.path, urlbase=array.urlbase) + + proxy = blosc2.open(urlpath, lazy=True, max_concurrency=3) + + assert isinstance(proxy, blosc2.Proxy) + assert isinstance(proxy.src, blosc2.C2Array) + assert proxy.src.max_concurrency == 3 + assert proxy.urlpath is None + + result = proxy[0:5, 0:10] + served = len(srv.log) + assert np.array_equal(result, data[0:5, 0:10]) + assert np.array_equal(proxy[0:5, 0:10], result) + assert len(srv.log) == served + + +def test_open_urlpath_lazy_persistent_cache(tmp_path, server, any_chunk_wants_blocks): + data = _incompressible((200, 200)) + array, srv = server(data, chunks=(100, 200), blocks=(10, 20)) + urlpath = blosc2.URLPath(array.path, urlbase=array.urlbase) + cache_storage = tmp_path / "cache" + + proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) + fetches = sum(endpoint == "fetch" for endpoint, _, _ in srv.log) + del proxy + + proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) + assert sum(endpoint == "fetch" for endpoint, _, _ in srv.log) == fetches + assert len(list(cache_storage.glob("*.b2nd"))) == 1 + + +def test_open_urlpath_lazy_uses_c2context_without_persisting_token(tmp_path, server): + token = "session=secret" + data = _incompressible((20, 20)) + array, _ = server(data, chunks=(10, 20), blocks=(5, 10), cookie=token) + urlpath = blosc2.URLPath(array.path) + cache_storage = tmp_path / "cache" + + with blosc2.c2context(urlbase=array.urlbase, auth_token=token): + proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + assert np.array_equal(proxy[0:5, 0:5], data[0:5, 0:5]) + assert proxy.schunk.meta["proxy-source"]["urlpath"][2] is None + + cache = next(cache_storage.glob("*.b2nd")) + reopened = blosc2.open(cache, mode="a") + assert np.array_equal(reopened[0:5, 0:5], data[0:5, 0:5]) + + +def test_open_urlpath_lazy_rebuilds_stale_cache(tmp_path, server, any_chunk_wants_blocks): + data = _incompressible((200, 200)) + array, srv = server(data, chunks=(100, 200), blocks=(10, 20)) + urlpath = blosc2.URLPath(array.path, urlbase=array.urlbase) + cache_storage = tmp_path / "cache" + + proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) + del proxy + + other = _incompressible((200, 200), seed=1) + _replace(srv, other, chunks=(100, 200), blocks=(10, 20)) + + proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + assert np.array_equal(proxy[0:5, 0:10], other[0:5, 0:10]) + + +def test_open_urlpath_cache_options_need_lazy(tmp_path, server): + data = _incompressible((20, 20)) + array, _ = server(data, chunks=(10, 20), blocks=(5, 10)) + urlpath = blosc2.URLPath(array.path, urlbase=array.urlbase) + + assert isinstance(blosc2.open(urlpath), blosc2.C2Array) + with pytest.raises(NotImplementedError, match=r"cache_storage.*lazy=True"): + blosc2.open(urlpath, cache_storage=tmp_path) + with pytest.raises(NotImplementedError, match=r"max_concurrency.*lazy=True"): + blosc2.open(urlpath, max_concurrency=2) + + def test_blocks_are_read_over_ranges(server, any_chunk_wants_blocks): data = _incompressible((200, 200)) array, srv = server(data, chunks=(100, 200), blocks=(10, 20)) From d7d20ccf9a8d9a678151bd5c23a8bbff9098d83d Mon Sep 17 00:00:00 2001 From: Francesc Alted Date: Thu, 3 Sep 2026 13:47:47 +0200 Subject: [PATCH 02/10] Optimize lazy HTTPS opening --- doc/guides/remote_arrays.md | 5 ++++ src/blosc2/proxy_source.py | 60 +++++++++++++++++++++++++++++++++++-- tests/test_fsspec.py | 46 ++++++++++++++++++++++++++-- 3 files changed, 105 insertions(+), 6 deletions(-) diff --git a/doc/guides/remote_arrays.md b/doc/guides/remote_arrays.md index d0e43b7b6..23eda6634 100644 --- a/doc/guides/remote_arrays.md +++ b/doc/guides/remote_arrays.md @@ -29,6 +29,11 @@ a[100:110, :50] # a NumPy array, fetched now `https://` means a plain web server — nginx, a CDN, an S3 website endpoint — anything that answers a `Range` request. A Caterva2 server is *not* reached that way: it names its datasets by root and path, so use {ref}`C2Array`. +A lazy HTTP(S) open takes its frame metadata and remote identity from the same +initial range response, using `ETag` when the server provides one and falling +back to `Last-Modified` and object size. Thus opening needs one network round +trip, and a persistent cache can still detect when the object is replaced. + ## The cache Wrap either of those in a {ref}`Proxy` and what you read is kept: diff --git a/src/blosc2/proxy_source.py b/src/blosc2/proxy_source.py index 8f825d5fb..957834bd7 100644 --- a/src/blosc2/proxy_source.py +++ b/src/blosc2/proxy_source.py @@ -1280,7 +1280,9 @@ def __init__(self, urlpath: str, max_concurrency: int = REMOTE_MAX_CONCURRENCY): fsspec = _import_fsspec(urlpath) fs, path = fsspec.url_to_fs(urlpath) - if fs.isdir(path): + protocols = (fs.protocol,) if isinstance(fs.protocol, str) else fs.protocol + self._http = bool({"http", "https"} & set(protocols)) + if not self._http and fs.isdir(path): raise NotImplementedError( f"{urlpath} is a directory (a sparse frame or a store), which cannot be read " "chunk by chunk; open it with cache_storage= instead" @@ -1290,15 +1292,67 @@ def __init__(self, urlpath: str, max_concurrency: int = REMOTE_MAX_CONCURRENCY): # has gone stale -- and chunk offsets from a replaced frame are garbage. # fsspec's own token, rather than a tuple of the metadata fields we guess # a backend exposes: memory:// has no mtime, which left it size-only. - self.stamp = fs.ukey(path) + # HTTPFileSystem.isdir() sends a GET before the range read below, making + # a lazy open pay two serial network round trips. Its ukey is only a + # hash of the URL and options, so it needs no request either. Capture + # ETag/Last-Modified from the first range response instead: one request + # supplies both the frame header and a stronger identity for the cache. + if self._http: + from fsspec.utils import tokenize + + self.stamp = tokenize(path, fs.kwargs, fs.protocol) + else: + self.stamp = fs.ukey(path) + self._capture_http_headers = self._http super().__init__(urlpath, max_concurrency) def read_range(self, offset: int, size: int) -> bytes: - data = self._fs.cat_file(self._path, start=offset, end=offset + size) + if self._capture_http_headers: + from fsspec.asyn import sync + + data, headers = sync( + self._fs.loop, + _http_cat_file_with_headers, + self._fs, + self._path, + offset, + offset + size, + ) + self.stamp = _http_stamp(self.stamp, headers) + self._capture_http_headers = False + else: + data = self._fs.cat_file(self._path, start=offset, end=offset + size) self.traffic.charge(len(data)) return data +async def _http_cat_file_with_headers(fs, url: str, start: int, end: int): + """HTTPFileSystem.cat_file(), returning the response headers as well.""" + kwargs = fs.kwargs.copy() + headers = kwargs.pop("headers", {}).copy() + headers["Range"] = await fs._process_limits(url, start, end) + kwargs["headers"] = headers + session = await fs.set_session() + async with session.get(fs.encode_url(url), **kwargs) as response: + data = await response.read() + fs._raise_not_found_for_status(response, url) + response_headers = {key.lower(): value for key, value in response.headers.items()} + return data, response_headers + + +def _http_stamp(url_stamp: str, headers: Mapping[str, str]) -> str: + """Combine a URL identity with the strongest validators on an HTTP response.""" + if etag := headers.get("etag"): + return f"{url_stamp}:etag:{etag}" + + modified = headers.get("last-modified", "") + content_range = headers.get("content-range", "") + size = content_range.rpartition("/")[2] if "/" in content_range else headers.get("content-length", "") + if modified or size: + return f"{url_stamp}:modified:{modified}:size:{size}" + return url_stamp + + def convert_dtype(dt: str | DTypeLike): """ Attempts to convert to blosc2.dtype (i.e. numpy dtype) diff --git a/tests/test_fsspec.py b/tests/test_fsspec.py index 2edac6d5a..cd0499f8e 100644 --- a/tests/test_fsspec.py +++ b/tests/test_fsspec.py @@ -8,6 +8,7 @@ import contextlib import functools +import hashlib import http.server import os import pathlib @@ -548,17 +549,53 @@ def test_http_url_is_read_through_fsspec(tmp_path): root.mkdir() blosc2.asarray(data, chunks=(50, 200), blocks=(10, 100), urlpath=str(root / "big.b2nd")) - with _ranged_server(root) as urlbase: + with _ranged_server(root) as (urlbase, requests): whole = blosc2.open(f"{urlbase}/big.b2nd") # fetched in one go, as s3:// is assert np.array_equal(whole[:], data) + requests.clear() lazy = blosc2.open(f"{urlbase}/big.b2nd", lazy=True, cache_storage=str(tmp_path / "cs")) assert isinstance(lazy, blosc2.Proxy) assert isinstance(lazy.src, blosc2.FsspecNDSource) - assert lazy.src.stamp is not None # so a cache of it can tell it has moved + assert ":etag:" in lazy.src.stamp + assert requests == ["bytes=0-8191"] # metadata and identity, one round trip assert np.array_equal(lazy[3:5, 100:120], data[3:5, 100:120]) +def test_http_lazy_cache_rebuilt_when_remote_changes(tmp_path): + pytest.importorskip("aiohttp") + path = tmp_path / "www" + path.mkdir() + frame = path / "changing.b2nd" + first = np.arange(40_000, dtype="i4").reshape(200, 200) + second = first + 1 + blosc2.asarray(first, chunks=(50, 200), blocks=(10, 100), urlpath=frame) + + with _ranged_server(path) as (urlbase, _): + url = f"{urlbase}/{frame.name}" + cache = tmp_path / "cache" + lazy = blosc2.open(url, lazy=True, cache_storage=cache) + assert np.array_equal(lazy[3:5, 100:120], first[3:5, 100:120]) + del lazy + + blosc2.asarray(second, chunks=(50, 200), blocks=(10, 100), urlpath=frame, mode="w") + lazy = blosc2.open(url, lazy=True, cache_storage=cache) + assert np.array_equal(lazy[3:5, 100:120], second[3:5, 100:120]) + + +def test_http_stamp_prefers_etag_and_falls_back_to_modified_size(): + stamp = blosc2.proxy_source._http_stamp( + "url", + {"etag": '"abc"', "last-modified": "yesterday", "content-range": "bytes 0-7/100"}, + ) + assert stamp == 'url:etag:"abc"' + + stamp = blosc2.proxy_source._http_stamp( + "url", {"last-modified": "yesterday", "content-range": "bytes 0-7/100"} + ) + assert stamp == "url:modified:yesterday:size:100" + + @contextlib.contextmanager def _ranged_server(root): """A web server over *root* that honours `Range`, which the stock one does not.""" @@ -571,6 +608,7 @@ def log_message(self, *args): def do_GET(self): span = self.headers.get("Range") + self.server.requests.append(span) if not span: return super().do_GET() body = (root / self.path.lstrip("/")).read_bytes() @@ -581,15 +619,17 @@ def do_GET(self): self.send_header("Content-Range", f"bytes {first}-{last}/{len(body)}") self.send_header("Accept-Ranges", "bytes") self.send_header("Content-Length", str(len(part))) + self.send_header("ETag", hashlib.sha256(body).hexdigest()) self.end_headers() self.wfile.write(part) return None handler = functools.partial(Ranged, directory=str(root)) server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler) + server.requests = [] threading.Thread(target=server.serve_forever, daemon=True).start() try: - yield f"http://127.0.0.1:{server.server_address[1]}" + yield f"http://127.0.0.1:{server.server_address[1]}", server.requests finally: server.shutdown() server.server_close() From 65fdd56a3570a1611036e36f40fed382b3de8e30 Mon Sep 17 00:00:00 2001 From: Francesc Alted Date: Thu, 3 Sep 2026 13:57:41 +0200 Subject: [PATCH 03/10] Clarify fsspec HTTPS example --- doc/getting_started/installation.rst | 9 +++++---- doc/guides/remote_arrays.md | 2 +- .../{s3-cat2-access.py => fsspec-cat2-access.py} | 14 +++++++------- pyproject.toml | 9 +++++---- 4 files changed, 18 insertions(+), 16 deletions(-) rename examples/{s3-cat2-access.py => fsspec-cat2-access.py} (88%) diff --git a/doc/getting_started/installation.rst b/doc/getting_started/installation.rst index b0b086a04..b4df7d7ed 100644 --- a/doc/getting_started/installation.rst +++ b/doc/getting_started/installation.rst @@ -42,10 +42,10 @@ grouped into *extras* that you opt into with the ``blosc2[extra]`` syntax: :doc:`../guides/parquet_to_blosc2`. * - ``fsspec`` - Reading and writing single-file containers through any `fsspec - `_ URL. The driver for each - protocol is a separate install (``s3fs`` for ``s3://``, ``gcsfs`` for - ``gs://``, ``adlfs`` for ``abfs://``...), and credentials are configured - through the driver, not through blosc2. + `_ URL. The HTTP(S) driver is + included. Other protocol drivers are separate installs (``s3fs`` for + ``s3://``, ``gcsfs`` for ``gs://``, ``adlfs`` for ``abfs://``...), and + credentials are configured through the driver, not through blosc2. Install one or more extras by listing them in brackets (quote the argument in shells like ``zsh`` that treat brackets specially): @@ -55,6 +55,7 @@ argument in shells like ``zsh`` that treat brackets specially): pip install "blosc2[tui]" # the b2view terminal browser pip install "blosc2[hires]" # b2view + its high-res view (h key) pip install "blosc2[parquet]" # the Parquet converter + pip install "blosc2[fsspec]" # fsspec URLs, including HTTP(S) pip install "blosc2[fsspec]" s3fs # fsspec URLs, plus the S3 driver pip install "blosc2[tui,parquet]" # several at once diff --git a/doc/guides/remote_arrays.md b/doc/guides/remote_arrays.md index 23eda6634..78497f093 100644 --- a/doc/guides/remote_arrays.md +++ b/doc/guides/remote_arrays.md @@ -251,6 +251,6 @@ Four things to get right: - {doc}`Tutorial 6 <../tutorials/06.remote_proxy>` — the same ground at a slower pace, with output. - `examples/ndarray/rw-fsspec.py` — every way of reading and writing an fsspec URL, runnable. -- `examples/s3-cat2-access.py` — the same dataset and cache API through HTTPS/fsspec and Caterva2, with timings. +- `examples/fsspec-cat2-access.py` — the same dataset and cache API through HTTPS/fsspec and Caterva2, with timings. - `examples/c2array-traffic.py` — what a remote slice costs in bytes, and what blocks and the cache save, runnable. - {ref}`C2Array`, {ref}`FsspecNDSource`, {ref}`ByteRangeNDSource`, {ref}`Proxy`, {ref}`Traffic` — the reference pages. diff --git a/examples/s3-cat2-access.py b/examples/fsspec-cat2-access.py similarity index 88% rename from examples/s3-cat2-access.py rename to examples/fsspec-cat2-access.py index c39733b46..94da9cc52 100644 --- a/examples/s3-cat2-access.py +++ b/examples/fsspec-cat2-access.py @@ -7,11 +7,11 @@ """Compare lazy access to the same array through fsspec and Caterva2. -The HTTPS path needs the fsspec HTTP dependencies. Install them with: +The HTTPS path needs the fsspec extra. Install it with: - pip install "blosc2[fsspec]" aiohttp + pip install "blosc2[fsspec]" -By default, caches are kept under ``./s3-cat2-cache``. Run the example again to +By default, caches are kept under ``./fsspec-cat2-cache``. Run the example again to see the first data access served by the cache left by the previous process. """ @@ -27,8 +27,8 @@ "@public/examples/cube-1k-1k-1k.b2nd", urlbase="https://cat2.cloud/demo", ) -# The same contents are published in this bucket with a ``-2`` suffix. -FSSPEC_URL = "https://blosc2.s3.us-west-001.backblazeb2.com/cube-1k-1k-1k-2.b2nd" +# The same contents are published in this Backblaze B2 bucket with a ``-2`` suffix. +FSSPEC_URL = "https://f001.backblazeb2.com/file/blosc2/cube-1k-1k-1k-2.b2nd" SLICE = np.s_[100:110, 200:300, 400:500] @@ -90,8 +90,8 @@ def main() -> None: parser.add_argument( "--cache-dir", type=Path, - default=Path("s3-cat2-cache"), - help="persistent cache root (default: ./s3-cat2-cache)", + default=Path("fsspec-cat2-cache"), + help="persistent cache root (default: ./fsspec-cat2-cache)", ) args = parser.parse_args() root = args.cache_dir diff --git a/pyproject.toml b/pyproject.toml index f52615b1c..b5703539d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,10 +59,11 @@ tui = ["textual", "textual-plotext"] # Adds the high-res 'h' view on top of [tui], rendering a real matplotlib image # (kitty/iTerm2/sixel, or half-cells elsewhere) — matplotlib is the heavy part. hires = ["blosc2[tui]", "textual-image", "matplotlib"] -# Read/write single-file containers through any fsspec URL (s3://, gs://, zip://, -# memory://...). The protocol backends (s3fs, gcsfs, adlfs...) are the caller's -# install: `pip install "blosc2[fsspec]" s3fs`. -fsspec = ["fsspec"] +# Read/write single-file containers through any fsspec URL (https://, s3://, +# gs://, zip://, memory://...). HTTP support is included; the other protocol +# backends (s3fs, gcsfs, adlfs...) are the caller's install: +# `pip install "blosc2[fsspec]" s3fs`. +fsspec = ["fsspec[http]"] [project.scripts] parquet-to-blosc2 = "blosc2.cli.parquet_to_blosc2:main" From 6c6e5a43b417d42c71375f254770103b7254514d Mon Sep 17 00:00:00 2001 From: Francesc Alted Date: Thu, 3 Sep 2026 18:59:06 +0200 Subject: [PATCH 04/10] Avoid redundant Caterva2 metadata refresh --- examples/fsspec-cat2-access.py | 6 +++++- src/blosc2/proxy.py | 15 +++++++++++---- src/blosc2/schunk.py | 13 +++++++++---- tests/ndarray/test_c2array_blocks.py | 9 +++++++-- 4 files changed, 32 insertions(+), 11 deletions(-) diff --git a/examples/fsspec-cat2-access.py b/examples/fsspec-cat2-access.py index 94da9cc52..8cb4e277b 100644 --- a/examples/fsspec-cat2-access.py +++ b/examples/fsspec-cat2-access.py @@ -23,12 +23,16 @@ import blosc2 +# Using the Caterva2 API CATERVA2_URL = blosc2.URLPath( "@public/examples/cube-1k-1k-1k.b2nd", urlbase="https://cat2.cloud/demo", ) +# ...and also using the fsspec path via fetch URL in Caterva2 +FSSPEC_URL = "https://cat2.cloud/demo/api/fetch/@public/examples/cube-1k-1k-1k.b2nd" # The same contents are published in this Backblaze B2 bucket with a ``-2`` suffix. -FSSPEC_URL = "https://f001.backblazeb2.com/file/blosc2/cube-1k-1k-1k-2.b2nd" +# FSSPEC_URL = "https://f001.backblazeb2.com/file/blosc2/cube-1k-1k-1k-2.b2nd" + SLICE = np.s_[100:110, 200:300, 400:500] diff --git a/src/blosc2/proxy.py b/src/blosc2/proxy.py index 5f7481c26..7d76d3378 100644 --- a/src/blosc2/proxy.py +++ b/src/blosc2/proxy.py @@ -68,7 +68,13 @@ class Proxy(blosc2.Operand): """ def __init__( - self, src: ProxySource or ProxyNDSource, urlpath: str | None = None, mode="a", **kwargs: dict + self, + src: ProxySource or ProxyNDSource, + urlpath: str | None = None, + mode="a", + *, + _refresh_source: bool = True, + **kwargs: dict, ): """ Create a new :ref:`Proxy` to serve as a cache to save accessed chunks locally. @@ -146,9 +152,10 @@ def __init__( # that has outlived someone else's writes would hand over a stamp the # cache still matches and a set of bytes it no longer does. Sources whose # bytes cannot move underneath them do not offer this and are not asked - refresh = getattr(self.src, "refresh_stamp", None) - if refresh is not None: - refresh() + if _refresh_source: + refresh = getattr(self.src, "refresh_stamp", None) + if refresh is not None: + refresh() if self._cache is None and mode == "a" and urlpath is not None and os.path.exists(urlpath): # Reuse the cache left by an earlier run: whatever was fetched then is diff --git a/src/blosc2/schunk.py b/src/blosc2/schunk.py index 402456fae..045c47ea5 100644 --- a/src/blosc2/schunk.py +++ b/src/blosc2/schunk.py @@ -1990,10 +1990,12 @@ def _lazy_fsspec_proxy( return _lazy_remote_proxy(src, urlpath, cache_storage) -def _lazy_remote_proxy(src, identity: str, cache_storage: str | pathlib.Path | None): +def _lazy_remote_proxy( + src, identity: str, cache_storage: str | pathlib.Path | None, *, source_fresh: bool = False +): """Wrap a remote source in a memory or persistent cache.""" if cache_storage is None: - return blosc2.Proxy(src) + return blosc2.Proxy(src, _refresh_source=not source_fresh) path = fsspec_cache_path(identity, cache_storage, ".b2nd") stamp = getattr(src, "stamp", None) @@ -2003,7 +2005,7 @@ def _lazy_remote_proxy(src, identity: str, cache_storage: str | pathlib.Path | N blosc2.remove_urlpath(path) # Proxy stamps the cache with src.stamp itself, and refuses one built against # other bytes; removing it above is what turns that refusal into a refetch - return blosc2.Proxy(src, urlpath=path, mode="a") + return blosc2.Proxy(src, urlpath=path, mode="a", _refresh_source=not source_fresh) def _open_c2_urlpath(urlpath: blosc2.URLPath, mode: str, offset: int, kwargs: dict): @@ -2031,7 +2033,10 @@ def _open_c2_urlpath(urlpath: blosc2.URLPath, mode: str, offset: int, kwargs: di if max_concurrency is not None: src.max_concurrency = max_concurrency identity = f"caterva2:{blosc2.c2array._server_url(src.urlbase, src.path)}" - return _lazy_remote_proxy(src, identity, cache_storage) + # C2Array's constructor has just read api/info. That response supplies both + # the geometry and the stamp against which the cache is checked, so asking + # for it again in Proxy.__init__ only adds a second serial round trip. + return _lazy_remote_proxy(src, identity, cache_storage, source_fresh=True) def _cache_stamp(path: str): diff --git a/tests/ndarray/test_c2array_blocks.py b/tests/ndarray/test_c2array_blocks.py index a5fce661d..0c57d3c7e 100644 --- a/tests/ndarray/test_c2array_blocks.py +++ b/tests/ndarray/test_c2array_blocks.py @@ -329,8 +329,10 @@ def test_open_urlpath_lazy_memory_cache(server, any_chunk_wants_blocks): array, srv = server(data, chunks=(100, 200), blocks=(10, 20)) urlpath = blosc2.URLPath(array.path, urlbase=array.urlbase) + srv.log.clear() proxy = blosc2.open(urlpath, lazy=True, max_concurrency=3) + assert [endpoint for endpoint, _, _ in srv.log] == ["info"] assert isinstance(proxy, blosc2.Proxy) assert isinstance(proxy.src, blosc2.C2Array) assert proxy.src.max_concurrency == 3 @@ -349,14 +351,17 @@ def test_open_urlpath_lazy_persistent_cache(tmp_path, server, any_chunk_wants_bl urlpath = blosc2.URLPath(array.path, urlbase=array.urlbase) cache_storage = tmp_path / "cache" + srv.log.clear() proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + assert [endpoint for endpoint, _, _ in srv.log] == ["info"] assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) - fetches = sum(endpoint == "fetch" for endpoint, _, _ in srv.log) del proxy + srv.log.clear() proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + assert [endpoint for endpoint, _, _ in srv.log] == ["info"] assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) - assert sum(endpoint == "fetch" for endpoint, _, _ in srv.log) == fetches + assert [endpoint for endpoint, _, _ in srv.log] == ["info"] assert len(list(cache_storage.glob("*.b2nd"))) == 1 From bca3cf413555eeb9b576317dc16a991b5000b293 Mon Sep 17 00:00:00 2001 From: Francesc Alted Date: Fri, 4 Sep 2026 07:41:57 +0200 Subject: [PATCH 05/10] Clarify and extend remote array caching --- doc/getting_started/installation.rst | 2 +- doc/guides/remote_arrays.md | 223 ++++++++++++--------------- doc/reference/c2array.rst | 18 ++- doc/reference/fsspecndsource.rst | 5 + doc/tutorials/06.remote_proxy.ipynb | 2 +- examples/fsspec-cat2-access.py | 8 +- examples/ndarray/rw-fsspec.py | 4 +- src/blosc2/c2array.py | 2 +- src/blosc2/proxy_source.py | 6 +- src/blosc2/schunk.py | 127 ++++++++++----- tests/ndarray/test_c2array_blocks.py | 41 +++-- tests/test_fsspec.py | 91 +++++++---- tests/test_fsspec_s3.py | 4 +- 13 files changed, 307 insertions(+), 226 deletions(-) diff --git a/doc/getting_started/installation.rst b/doc/getting_started/installation.rst index b4df7d7ed..36f3213c5 100644 --- a/doc/getting_started/installation.rst +++ b/doc/getting_started/installation.rst @@ -60,7 +60,7 @@ argument in shells like ``zsh`` that treat brackets specially): pip install "blosc2[tui,parquet]" # several at once With the ``fsspec`` extra, :func:`blosc2.open` accepts any fsspec URL, chained -ones included, and reads it whole, through a local cache (``cache_storage=``), or +ones included, and reads it whole, through a local cache (``cache_dir=``), or by fetching only the chunks and blocks a slice touches (``lazy=True``); see :func:`blosc2.open` and :ref:`FsspecNDSource` for what each mode supports. ``examples/ndarray/rw-fsspec.py`` walks through all three plus the write side, diff --git a/doc/guides/remote_arrays.md b/doc/guides/remote_arrays.md index 78497f093..4010b67fa 100644 --- a/doc/guides/remote_arrays.md +++ b/doc/guides/remote_arrays.md @@ -1,163 +1,156 @@ # Working with Remote Arrays -A Blosc2 array that lives on a server does not have to be downloaded to be used. Blosc2 opens it where it is, fetches only the pieces a slice touches, and keeps those in a local cache so the next run starts from them. +Blosc2 can open an array without downloading it first. Metadata is read at open time; array data is fetched only when a slice needs it and is then kept in a local cache. -## Three ways in +## Choose a remote route -| Where the array lives | How to open it | -|---|---| -| Any URL fsspec reaches — `s3://`, `gs://`, `https://`, `zip://`… | `blosc2.open(url, lazy=True)` | -| A [Caterva2](https://ironarray.io/caterva2) server | `blosc2.open(blosc2.URLPath(path, urlbase=...), lazy=True)` | -| Anything else | A `read_range()` of your own — see [Your own transport](#your-own-transport) | +The argument passed to {func}`blosc2.open` selects the route: + +| Argument | Route | What it names | +|---|---|---| +| A URL string such as `s3://...` or `https://...` | fsspec | A byte-addressable, standalone `.b2nd` file | +| A {ref}`URLPath` | Caterva2 | One array-like dataset on a Caterva2 server | ```python import blosc2 -# An object store, a web server, a zip on either of them +# fsspec: an object store or plain web server a = blosc2.open("s3://bucket/big.b2nd", lazy=True) -# A Caterva2 server; add lazy=True for an automatic Proxy cache +# Caterva2: a dataset identified by root and path b = blosc2.open( blosc2.URLPath( - "@public/examples/lung-jpeg2000_10x.b2nd", urlbase="https://cat2.cloud/demo" - ) + "@public/examples/lung-jpeg2000_10x.b2nd", + urlbase="https://cat2.cloud/demo", + ), + lazy=True, ) -a.shape, a.dtype # metadata only; nothing was downloaded -a[100:110, :50] # a NumPy array, fetched now +a.shape, a.dtype # metadata is available immediately +a[100:110, :50] # data is fetched now ``` -`https://` means a plain web server — nginx, a CDN, an S3 website endpoint — anything that answers a `Range` request. A Caterva2 server is *not* reached that way: it names its datasets by root and path, so use {ref}`C2Array`. +A `URLPath` always means Caterva2. If its `urlbase` is omitted, the server comes from {func}`blosc2.c2context` or `BLOSC_C2URLBASE`. Other transports can be added with a custom {ref}`ByteRangeNDSource`; see [Use your own transport](#use-your-own-transport). -A lazy HTTP(S) open takes its frame metadata and remote identity from the same -initial range response, using `ETag` when the server provides one and falling -back to `Last-Modified` and object size. Thus opening needs one network round -trip, and a persistent cache can still detect when the object is replaced. +### What each route supports -## The cache +Both routes return a {ref}`Proxy` when opened with `lazy=True`, so slicing and caching work the same way. Their sources differ: -Wrap either of those in a {ref}`Proxy` and what you read is kept: +| Remote object | fsspec URL | Caterva2 `URLPath` | +|---|---|---| +| Standalone contiguous `.b2nd` | Yes | Yes | +| HDF5 dataset | No | Yes | +| NDArray leaf inside `.b2z` | No | Yes | +| Lazy or computed array | No | Yes | +| Whole `.b2z` `TreeStore` or `DictStore` | No | No; open one array-like leaf | -```python -p = blosc2.Proxy(b) # cache in memory, gone when the proxy is -p[10:12, 500:600] # fetched from the server, and kept -p[10:12, 500:600] # read from the cache, no request at all -``` +fsspec supplies byte ranges. Python-Blosc2 parses the `.b2nd` frame to discover its geometry and chunk offsets, making this route direct and efficient for standalone arrays. + +Caterva2 understands dataset paths, array metadata, and slicing. It can therefore expose array-like data that is not stored as a standalone Blosc2 frame, as well as apply authentication or server-side computation. Use Caterva2's navigation API to find a leaf in a remote hierarchy, then open that leaf with a `URLPath`. + +`lazy=True` changes when data is fetched; it does not expand the formats supported by either route. + +## Choose a cache -Where that cache lives is yours to choose, and it is the one decision to make here. Say nothing and it is memory: fast, and it dies with the proxy, which is all a single process reading a slice twice needs. Name a file with `urlpath=` and the cache outlives the run: +Every lazy open creates a cache. By default it lives in memory and disappears with the proxy: ```python -p = blosc2.Proxy(b, urlpath="lung-cache.b2nd", mode="a") -p[10:12, 500:600] # fetched from the server, and written to lung-cache.b2nd +a = blosc2.open("s3://bucket/big.b2nd", lazy=True) +a[10:12, 500:600] # fetched and cached +a[10:12, 500:600] # served from memory ``` -That file is an ordinary Blosc2 array holding only the pieces you touched — a few hundred bytes for a freshly opened proxy over a 64 MB dataset, growing as you read. It is a normal `.b2nd`: copy it, ship it, open it with {func}`blosc2.open`. With `mode="a"` a later run picks up where the last one left off. - -{func}`blosc2.open` builds the proxy for either kind of remote source and offers -the same choice under another name — `cache_storage=` for a cache on disk, -nothing for one in memory: +Set `cache_dir` to let Blosc2 manage a cache file inside a directory: ```python url = "s3://bucket/big.b2nd" -# First run: the slice is fetched, and lands under ./b2cache dir -a = blosc2.open(url, lazy=True, cache_storage="./b2cache") -a[100:110, :50] +a = blosc2.open(url, lazy=True, cache_dir="./b2cache") +a[100:110, :50] # fetched and stored under ./b2cache -# A later run, a different process: same call, served from ./b2cache -a = blosc2.open(url, lazy=True, cache_storage="./b2cache") +# A later process can reuse the same cache. +a = blosc2.open(url, lazy=True, cache_dir="./b2cache") a[100:110, :50] # no request ``` -The same interface works for Caterva2: +Use `cache_path` instead when the cache should have an exact filename: ```python -url = blosc2.URLPath("@personal/run.b2nd") - -with blosc2.c2context( - urlbase="https://cat2.cloud/demo", - username="me@example.com", - password="secret", -): - a = blosc2.open(url, lazy=True, cache_storage="./b2cache") - a[100:110, :50] +a = blosc2.open(url, lazy=True, cache_path="big-cache.b2nd") ``` -For authenticated Caterva2 datasets, `cache_storage` must be private to the -current user. Applications serving multiple users must use a separate cache -directory for each user; sharing one between users is not supported. Reopen a -private cache inside an equivalent authenticated {func}`c2context`. +In both cases, the cache is an ordinary `.b2nd` array that starts small and grows as regions are read. `cache_dir` and `cache_path` are mutually exclusive. -## Only what a slice touches +Authenticated Caterva2 caches must be private to one user. Reopen them under an equivalent authenticated {func}`blosc2.c2context`; do not share a cache directory between users. -A chunk is the unit a container is compressed in, and it can be several megabytes. Fetching a whole one to read a corner of it is most of the cost of a remote read, so Blosc2 fetches **blocks** — the smaller pieces a chunk is built from — whenever a slice lands in a small part of a large chunk. +## Only what a slice touches -You do not ask for this; it happens when it pays. For example: +Blosc2 arrays are compressed in chunks, which are divided into smaller blocks. For a small slice, fetching only its blocks can avoid transferring most of a large chunk. -- On S3, block reads are **5–17x faster** on arrays with multi-megabyte chunks, and **2–5x** on 1 MB ones. -- On cat2.cloud's `kevlar-tomo.b2nd`, a corner slice costs **0.031 MB instead of 2.723 MB**, and a slice touching ten chunks takes **0.14 s against 1.01 s**. +![A proxy fetches missing regions from the remote array into its local cache. The fetch method returns the cache container, while indexing returns only the requested values.](../tutorials/images/remote_proxy.png) -It is never a loss. A slice wanting more than half a chunk's blocks is wanting the chunk, and a fetch that would skip too little to pay for the extra round trip is made whole — both answered from metadata already in hand, before anything is read. Where blocks are not available the read falls back to whole chunks by itself: that happens for a dataset a Caterva2 server *computes* rather than stores (a lazy expression, an HDF5 leaf, a `.b2z` member), and for a server that stops honouring ranges. +Purple regions are cached; red regions are still remote. The grid is schematic: where byte ranges are available, the fetched regions can be blocks within a chunk. `fetch()` fills and returns the cache container, whereas indexing returns only the requested values. -Fetches also overlap: a lazy proxy runs 8 at a time by default. Pass `max_concurrency=1` for a local protocol with no latency to hide. +The proxy chooses blocks or whole chunks automatically. It fetches a whole chunk when most of its blocks are needed or when the source cannot expose block ranges, as with computed Caterva2 datasets. Independent reads overlap, with up to eight concurrent requests by default; use `max_concurrency=1` when concurrency does not help. -A step other than 1 needs a proxy — a bare {ref}`C2Array` refuses one. Through a proxy it is placed on the block grid like any other key: `p[::2]` reads the blocks holding the coordinates it selects and no others, and `[::-1]` costs what its forward twin does. What that saves is `min(step, block extent along that axis)`, so it is nothing where blocks already span the axis whole — a step along the last dimension, usually — and the step's own factor where they do not. On `kevlar-tomo.b2nd`, whose blocks are one row deep, `[::2]` halves the read and `[::5]` cuts it fivefold. +Stepped slices also use the block grid. For example, `p[::5]` can reduce transfers along an axis whose blocks do not already span that axis. A bare {ref}`C2Array` does not accept stepped slices; its proxy does. -### Seeing byte savings +### Measure network traffic -Wall time will not show you any of this: on a fast link a block read and a whole-chunk read take about as long and differ by the compression ratio in *bytes*. Bytes are also what a metered link and a shared server uplink run out of, so they are counted for you. {ref}`C2Array` and {ref}`Proxy` each carry a {ref}`Traffic` under `traffic` — cumulative requests and bytes, tallied at the transport, so the frame index and block offsets are in it too: +{ref}`C2Array` and remote {ref}`Proxy` objects expose cumulative request and byte counts through {ref}`Traffic`: ```python -b = blosc2.C2Array( - "@public/examples/kevlar-tomo.b2nd", urlbase="https://cat2.cloud/demo" +source = blosc2.C2Array( + "@public/examples/kevlar-tomo.b2nd", + urlbase="https://cat2.cloud/demo", ) -p = blosc2.Proxy(b) +p = blosc2.Proxy(source) p.traffic.reset() corner = p[0, :100, :100] -print(p.traffic) # Traffic(requests=4, nbytes=57767) +print(p.traffic) # requests and bytes fetched p.traffic.reset() -p[0, :100, :100] # the same slice, from the cache +p[0, :100, :100] print(p.traffic) # Traffic(requests=0, nbytes=0) ``` -Take two readings and subtract, or `reset()` between them. `Proxy.traffic` is `None` over a local array — nothing crosses a wire there, and a zero would say the traffic was free rather than that it was never measured. `examples/c2array-traffic.py` runs the whole comparison against cat2.cloud's `kevlar-tomo.b2nd`: a 100x100 corner costs 0.055 MB against 1.296 MB for the chunk holding it — 23.5x — and nothing at all on the second read. +Use `reset()` or subtract two readings to measure one operation. `Proxy.traffic` is `None` for a local source because no network transport exists. + +`examples/c2array-traffic.py` compares block, chunk, and cached reads against a live Caterva2 dataset. -## Scattered points +## Retrieve scattered points -A list of coordinates, or a boolean mask, is not a box — but every point it picks still lives in exactly one block, so it is placed on the block grid as exactly as a slice is: +A proxy maps coordinate arrays and boolean masks to the blocks that contain their selected points: ```python -p[rows, :100] # rows is an array of three indices: three blocks, not three chunks -p[mask] # a mask picks coordinates too, and costs the same +p[rows, :100] +p[mask] ``` -Nine scattered points of a 900³ array cost **236 KB in 19 requests** through a proxy, against 1.81 MB for the chunks holding them. +For Caterva2, a bare {ref}`C2Array` can be substantially more efficient: it sends the coordinates to the server, which returns only the selected values. Prefer direct `C2Array` indexing for sparse, one-off point retrieval; prefer a proxy when reuse through a local cache matters. -However, a {ref}`C2Array` does better with no proxy at all: the coordinates go to the server, which gathers the points and sends back those alone — **271 bytes in one request** for the same nine. When you need efficient scattered retrievals, C2Array+Caterva2 is your best friend. +## Handle remote changes -## When the remote changes underneath +A persistent cache records the source identity when one is available. On a later `blosc2.open()` with the same `cache_dir` or `cache_path`, a mismatched cache is discarded and rebuilt automatically. -A cache is only good while the bytes it was filled from are still there. Sources that can name their bytes — an fsspec URL by its token, a Caterva2 array by an identifier the server keeps — are checked against what the cache recorded: +When constructing a proxy directly in append mode, a mismatch is reported instead: ```python -p = blosc2.Proxy(src, urlpath="cache.b2nd", mode="a") -# ValueError: the cache at cache.b2nd was built against different remote bytes; -# pass mode='w' to fetch them anew +p = blosc2.Proxy(source, urlpath="cache.b2nd", mode="a") +# ValueError if cache.b2nd belongs to different remote bytes ``` -`mode="w"` starts the cache empty and refetches. For a source that cannot name its bytes, the cache is adopted on geometry alone — same shape, dtype and partitioning — so an array rewritten in place while its geometry stayed the same is served from the cache as it was. Use `mode="w"` when that is a possibility. +Use `mode="w"` to start that cache again. If a source cannot provide an identity, compatibility is checked only from shape, dtype, chunks, and blocks. Use a fresh cache when such a source may have changed without changing its geometry. -## Filling an array from several writers +## Fill a Caterva2 array concurrently -A Caterva2 array can be *written*, one chunk at a time, by as many processes as it has chunks. Lay the array out empty first — {func}`blosc2.uninit` writes a couple of hundred bytes whatever the shape — upload it to the server, then have each writer post the chunks it owns: +Several writers can fill one Caterva2 array when each chunk is written at most once. First create and upload an uninitialized array with its final geometry: ```python import blosc2 import numpy as np -# Once, before the writers start: an empty array of the final geometry blosc2.uninit( (1_000_000,), dtype=np.float64, @@ -167,50 +160,39 @@ blosc2.uninit( ) ``` -Upload it with the client that comes with Caterva2: - ```sh cat2-client upload run.b2nd @personal/run.b2nd ``` -Then each writer opens it and posts its own chunks: +Each writer compresses and posts the chunks it owns: ```python import math -import blosc2 - a = blosc2.C2Array("@personal/run.b2nd", urlbase="https://cat2.cloud/demo") -itemsize = a.dtype.itemsize chunk = blosc2.compress2( - data, typesize=itemsize, blocksize=math.prod(a.blocks) * itemsize + data, + typesize=a.dtype.itemsize, + blocksize=math.prod(a.blocks) * a.dtype.itemsize, ) -a.update_chunk(nchunk, chunk) -``` - -Each slot is written once. A second write to the same slot raises {class}`blosc2.ChunkAlreadyWritten`, and that refusal is the whole of the coordination — two writers that both think they own a chunk are sorted out by the array, with no lease, lock or registry between them. The loser drops its chunk and moves on: -```python try: a.update_chunk(nchunk, chunk) except blosc2.ChunkAlreadyWritten: - pass # someone else got there first + pass # another writer completed this slot ``` -Writing into an empty slot appends to the file and moves no other chunk, which is what makes a fill cheap and lets a reader follow one without its cached positions going wrong. {meth}`C2Array.written_chunks() ` says how far it has got, straight out of the file's own index — no endpoint of its own, about 2.5 ms over HTTP: +The server serializes updates, and {meth}`C2Array.written_chunks() ` reports progress from the array's index: ```python -written = a.written_chunks() # one bool per chunk -print(f"{written.sum()}/{written.size} chunks in") +written = a.written_chunks() for nchunk in np.flatnonzero(~written): - ... # the work still to do, after a crash + ... # chunks still missing after a restart ``` -What this buys: the server serializes the writes themselves, so what overlaps is the round trip — which over a network is nearly all of the cost. Against a real server, a fill went from **244 ms per chunk serially to 32 ms with 8 writers, 7.6x**. Over loopback, where there is no round trip to hide, it is 1.0x. +## Use your own transport -## Your own transport - -If your frames live somewhere fsspec does not reach — per-request credentials, a signing proxy, a database column, an in-house gateway — supply one method and you get everything above: +Subclass {ref}`ByteRangeNDSource` when the frame lives behind a transport that fsspec cannot use: ```python import boto3 @@ -219,18 +201,18 @@ import blosc2 class S3Source(blosc2.ByteRangeNDSource): def __init__(self, bucket, key): - self._s3 = boto3.client("s3") - self._bucket, self._key = bucket, key - self.stamp = self._s3.head_object(Bucket=bucket, Key=key)["ETag"] + self.s3 = boto3.client("s3") + self.bucket, self.key = bucket, key + self.stamp = self.s3.head_object(Bucket=bucket, Key=key)["ETag"] super().__init__(f"s3://{bucket}/{key}") def read_range(self, offset, size): - answer = self._s3.get_object( - Bucket=self._bucket, - Key=self._key, + response = self.s3.get_object( + Bucket=self.bucket, + Key=self.key, Range=f"bytes={offset}-{offset + size - 1}", ) - data = answer["Body"].read() + data = response["Body"].read() self.traffic.charge(len(data)) return data @@ -238,19 +220,14 @@ class S3Source(blosc2.ByteRangeNDSource): a = blosc2.Proxy(S3Source("bucket", "big.b2nd"), urlpath="cache.b2nd", mode="a") ``` -(For plain S3 you would just use `blosc2.open("s3://bucket/big.b2nd", lazy=True)`; this is the shape of the thing.) - -Four things to get right: +Initialize the transport before `super().__init__()`, because the base constructor immediately reads the frame header. Make `read_range()` thread-safe, set `stamp` so persistent caches can detect changes, and charge the bytes read so traffic measurements remain accurate. -- **Set up the transport before `super().__init__()`.** The base constructor calls `read_range()` straight away to read the file's header. -- **`read_range()` must be thread-safe.** It is called from a thread pool so fetches can overlap. A boto3 *client* is fine; a `Session` or resource is not. -- **Set `stamp` if you can.** It is what lets a cache tell that the remote has changed. Without it the cache is kept on geometry alone. -- **Charge what you read.** End `read_range()` with `self.traffic.charge(len(data))` and your source is counted like the built-in ones — see [Seeing byte savings](#seeing-byte-savings). Skip it and `traffic` reads zero forever, which looks like a free transport rather than an uncounted one. +For ordinary S3 access, use `blosc2.open("s3://bucket/big.b2nd", lazy=True)`; the custom class only illustrates the transport contract. ## See also -- {doc}`Tutorial 6 <../tutorials/06.remote_proxy>` — the same ground at a slower pace, with output. -- `examples/ndarray/rw-fsspec.py` — every way of reading and writing an fsspec URL, runnable. -- `examples/fsspec-cat2-access.py` — the same dataset and cache API through HTTPS/fsspec and Caterva2, with timings. -- `examples/c2array-traffic.py` — what a remote slice costs in bytes, and what blocks and the cache save, runnable. -- {ref}`C2Array`, {ref}`FsspecNDSource`, {ref}`ByteRangeNDSource`, {ref}`Proxy`, {ref}`Traffic` — the reference pages. +- {doc}`Tutorial 6 <../tutorials/06.remote_proxy>` — a step-by-step introduction with output. +- `examples/ndarray/rw-fsspec.py` — fsspec reading and writing examples. +- `examples/fsspec-cat2-access.py` — one dataset and cache through fsspec and Caterva2. +- `examples/c2array-traffic.py` — block, chunk, and cached transfer sizes. +- {ref}`C2Array`, {ref}`FsspecNDSource`, {ref}`ByteRangeNDSource`, {ref}`Proxy`, and {ref}`Traffic` — API reference pages. diff --git a/doc/reference/c2array.rst b/doc/reference/c2array.rst index 9f8caaef2..2085ce11e 100644 --- a/doc/reference/c2array.rst +++ b/doc/reference/c2array.rst @@ -3,16 +3,26 @@ C2Array ======= -This is a class for remote arrays. This kind of array can also work as operand on a LazyExpr, LazyUDF or reduction. +This is a class for one array-like dataset addressed through a Caterva2 server. +The dataset may be a standalone ``.b2nd`` array, an HDF5 dataset, an NDArray +leaf inside a ``.b2z`` store, or a lazy/computed array. A ``C2Array`` does not +represent or navigate a whole remote ``TreeStore`` or ``DictStore``; use +Caterva2 to select a leaf and open that leaf's path. This kind of array can also +work as an operand on a LazyExpr, LazyUDF or reduction. :ref:`URLPath` is +Caterva2-only, including when its ``urlbase`` is omitted and inherited from +:func:`blosc2.c2context`. + +For a comparison with byte-oriented fsspec access, see +:doc:`Working with Remote Arrays <../guides/remote_arrays>`. Wrapped in a :ref:`Proxy`, a stored remote array is read at block granularity: the proxy asks for the blocks a slice touches rather than the chunks they live in, which for a multi-megabyte chunk is a small fraction of the bytes. That rests on the server serving the dataset from a file, ``Range`` header and auth cookie both honoured; a dataset it computes instead (a lazy expression, an -HDF5 leaf) is fetched a whole chunk at a time, as everything was before. Which -one this is takes at most one request to find out, and is decided once -- -:meth:`C2Array.block_source` is what answers it. +HDF5 leaf, or a ``.b2z`` member) is fetched a whole chunk at a time, as +everything was before. Which one this is takes at most one request to find out, +and is decided once -- :meth:`C2Array.block_source` is what answers it. A stored remote array can also be *filled*, by as many writers at once as it has chunks. The array is laid out first -- ``blosc2.uninit`` writes a couple of diff --git a/doc/reference/fsspecndsource.rst b/doc/reference/fsspecndsource.rst index deb2a6fb6..8b53dfff0 100644 --- a/doc/reference/fsspecndsource.rst +++ b/doc/reference/fsspecndsource.rst @@ -7,6 +7,11 @@ A :ref:`ByteRangeNDSource` that serves the chunks of a Blosc2 frame living behind an fsspec URL, reading each one with a range request instead of transferring the whole container. Everything about the frame format, block granularity included, lives in the base class; this adds the fsspec transport. +The URL must name a standalone, contiguous ``.b2nd`` NDArray frame. It cannot +name an HDF5 dataset, a member inside a ``.b2z`` store, a sparse directory +container, or a computed array: fsspec provides bytes, not dataset semantics. +For the Caterva2 alternative and a capability comparison, see +:doc:`Working with Remote Arrays <../guides/remote_arrays>`. For other sources, see :ref:`ProxyNDSource` and :ref:`ProxySource`. ``examples/ndarray/rw-fsspec.py`` is a runnable walkthrough of this and the diff --git a/doc/tutorials/06.remote_proxy.ipynb b/doc/tutorials/06.remote_proxy.ipynb index 9c1cf8a98..41d8f6bc8 100644 --- a/doc/tutorials/06.remote_proxy.ipynb +++ b/doc/tutorials/06.remote_proxy.ipynb @@ -38,7 +38,7 @@ "metadata": {}, "source": [ "## ``C2Array`` class\n", - "Before we look at proxies, it is first necessary to understand how to use Blosc2 to work with remote data, via the ``C2Array`` class. The class implements a (limited) version of the NDArray interface of which we have already seen a lot in previous tutorials. However, it is really a local pointer to a remote array (stored e.g. on a remote server). This means that we can refer to the data, access certain attribute information about it, download portions of the data and even define it in computational expressions, without having to download the entire array into local memory or disk. This is particularly useful when working with large datasets that cannot fit into memory or would take far too long to transfer over the network.\n", + "Before we look at proxies, it is first necessary to understand how to use Blosc2 to work with remote data, via the ``C2Array`` class. The class implements a (limited) version of the NDArray interface of which we have already seen a lot in previous tutorials. However, it is really a local pointer to a remote array (stored e.g. on a remote server). This means that we can refer to the data, access certain attribute information about it, download portions of the data and even define it in computational expressions, without having to download the entire array into local memory or disk. This is particularly useful when working with large datasets that cannot fit into memory or would take far too long to transfer over the network. The [Working with Remote Arrays](https://www.blosc.org/python-blosc2/guides/remote_arrays.html#choosing-between-fsspec-and-caterva2) guide explains when to use Caterva2's semantic dataset route instead of a byte-oriented fsspec URL.\n", "\n", "However, one limitation of this approach is that every time one wants to download a slice of the dataset, the data is fetched over the network - even if the same slice has been downloaded before. This can lead to inefficiencies, especially when working with large datasets or when the same data is accessed multiple times. Proxies offer a solution to this, whilst still preserving the low storage requirements of the ``C2Array`` class.\n", "\n", diff --git a/examples/fsspec-cat2-access.py b/examples/fsspec-cat2-access.py index 8cb4e277b..78e21cfbf 100644 --- a/examples/fsspec-cat2-access.py +++ b/examples/fsspec-cat2-access.py @@ -46,11 +46,11 @@ def size_text(size: int) -> str: return f"{size / 2**20:.3f} MiB" -def benchmark(label: str, urlpath, cache_storage: Path) -> np.ndarray: - cache_existed = cache_storage.is_dir() and any(cache_storage.glob("*.b2nd")) +def benchmark(label: str, urlpath, cache_dir: Path) -> np.ndarray: + cache_existed = cache_dir.is_dir() and any(cache_dir.glob("*.b2nd")) start = perf_counter() - array = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + array = blosc2.open(urlpath, lazy=True, cache_dir=cache_dir) open_time = perf_counter() - start metadata = (array.shape, array.dtype, array.chunks, array.blocks) @@ -67,7 +67,7 @@ def benchmark(label: str, urlpath, cache_storage: Path) -> np.ndarray: # that cached data survives the Proxy object, not merely one array access. del array start = perf_counter() - reopened = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + reopened = blosc2.open(urlpath, lazy=True, cache_dir=cache_dir) reopen_time = perf_counter() - start reopened.traffic.reset() diff --git a/examples/ndarray/rw-fsspec.py b/examples/ndarray/rw-fsspec.py index c7aa286ca..1d7c09da0 100644 --- a/examples/ndarray/rw-fsspec.py +++ b/examples/ndarray/rw-fsspec.py @@ -43,7 +43,7 @@ # starts from the copy that is already there. Cached copies are checked # against the remote on every open, so a replaced array is never served # from a stale cache. - c = blosc2.open(urlpath, cache_storage=cachedir, mmap_mode="r") + c = blosc2.open(urlpath, cache_dir=cachedir, mmap_mode="r") print(f"read cached: {c.shape} (mmapped from {cachedir})") np.testing.assert_array_equal(c[:], a[:]) @@ -51,7 +51,7 @@ # is and each slice fetches only what it touches -- the chunks it lands in, # or just the blocks inside them when the chunks are large enough for that # to pay. This is what you want for an array too big to download. - d = blosc2.open(urlpath, lazy=True, cache_storage=cachedir) + d = blosc2.open(urlpath, lazy=True, cache_dir=cachedir) print(f"read lazy: {type(d).__name__} {d.shape} {d.dtype}") # Only the two chunks covering rows 15..25 are fetched here diff --git a/src/blosc2/c2array.py b/src/blosc2/c2array.py index a2745a155..9286894b9 100644 --- a/src/blosc2/c2array.py +++ b/src/blosc2/c2array.py @@ -1672,7 +1672,7 @@ def __init__(self, path: str, /, urlbase: str | None = None, auth_token: str | N Passing this object to :func:`blosc2.open` returns a :ref:`C2Array`. With ``lazy=True`` it instead returns a :ref:`Proxy`, using an in-memory cache - by default or a persistent cache when ``cache_storage`` is provided. + by default or a persistent cache when ``cache_dir`` is provided. Authenticated users sharing a machine must use separate cache directories. The parameters are the same as for the :meth:`C2Array.__init__`. diff --git a/src/blosc2/proxy_source.py b/src/blosc2/proxy_source.py index 957834bd7..a775d1629 100644 --- a/src/blosc2/proxy_source.py +++ b/src/blosc2/proxy_source.py @@ -734,7 +734,7 @@ def __init__( except KeyError: raise NotImplementedError( f"{urlpath} has no b2nd metalayer, so it is a plain SChunk rather than an " - "NDArray; read it whole or with cache_storage= instead" + "NDArray; read it whole or with cache_dir= instead" ) from None if dtype_format != 0: raise NotImplementedError(f"unsupported dtype format {dtype_format} in {urlpath}") @@ -1262,7 +1262,7 @@ class FsspecNDSource(ByteRangeNDSource): This is what ``blosc2.open(url, lazy=True)`` builds; wrap it in a :ref:`Proxy` by hand when the cache belongs at a path of your choosing - rather than inside ``cache_storage``:: + rather than inside ``cache_dir``:: src = blosc2.FsspecNDSource("s3://bucket/big.b2nd") a = blosc2.Proxy(src, urlpath="big-cache.b2nd", mode="a") @@ -1285,7 +1285,7 @@ def __init__(self, urlpath: str, max_concurrency: int = REMOTE_MAX_CONCURRENCY): if not self._http and fs.isdir(path): raise NotImplementedError( f"{urlpath} is a directory (a sparse frame or a store), which cannot be read " - "chunk by chunk; open it with cache_storage= instead" + "chunk by chunk; open it with cache_dir= instead" ) self._fs, self._path = fs, path # Identifies the remote bytes, so a cache built against them can tell it diff --git a/src/blosc2/schunk.py b/src/blosc2/schunk.py index 045c47ea5..ac23c6a16 100644 --- a/src/blosc2/schunk.py +++ b/src/blosc2/schunk.py @@ -10,6 +10,7 @@ import builtins import os import pathlib +import warnings import weakref import zipfile from collections import namedtuple @@ -1975,29 +1976,62 @@ def _finalize_special_open(special, urlpath, mode): return special +def _remote_cache_options(kwargs: dict) -> tuple[str | pathlib.Path | None, str | pathlib.Path | None]: + """Pop the public remote-cache options, including the deprecated alias.""" + legacy_present = "cache_storage" in kwargs + cache_storage = kwargs.pop("cache_storage", None) + cache_dir = kwargs.pop("cache_dir", None) + cache_path = kwargs.pop("cache_path", None) + + if legacy_present: + warnings.warn( + "cache_storage is deprecated; use cache_dir instead", + DeprecationWarning, + stacklevel=4, + ) + + selected = [value for value in (cache_storage, cache_dir, cache_path) if value is not None] + if len(selected) > 1: + raise ValueError("cache_storage, cache_dir, and cache_path are mutually exclusive") + return (cache_dir if cache_dir is not None else cache_storage), cache_path + + def _lazy_fsspec_proxy( - urlpath: str, cache_storage: str | pathlib.Path | None, max_concurrency: int | None = None + urlpath: str, + cache_dir: str | pathlib.Path | None, + cache_path: str | pathlib.Path | None, + max_concurrency: int | None = None, ): """Wrap a remote frame in a Proxy that fetches chunks on demand. - Without `cache_storage` the fetched chunks live in memory and die with the - proxy; with it they go to a container under that directory, so a later run - starts from what this one pulled. + Without a cache location the fetched chunks live in memory and die with the + proxy. Otherwise they go to `cache_path`, or to a derived name under + `cache_dir`, so a later run starts from what this one pulled. """ # None leaves the default where it belongs, on the source itself kwargs = {} if max_concurrency is None else {"max_concurrency": max_concurrency} src = blosc2.FsspecNDSource(urlpath, **kwargs) - return _lazy_remote_proxy(src, urlpath, cache_storage) + return _lazy_remote_proxy(src, urlpath, cache_dir, cache_path) def _lazy_remote_proxy( - src, identity: str, cache_storage: str | pathlib.Path | None, *, source_fresh: bool = False + src, + identity: str, + cache_dir: str | pathlib.Path | None, + cache_path: str | pathlib.Path | None, + *, + source_fresh: bool = False, ): """Wrap a remote source in a memory or persistent cache.""" - if cache_storage is None: + if cache_dir is None and cache_path is None: return blosc2.Proxy(src, _refresh_source=not source_fresh) - path = fsspec_cache_path(identity, cache_storage, ".b2nd") + if cache_path is not None: + path = os.fspath(cache_path) + if os.path.isdir(path): + raise ValueError("cache_path must name a file, not a directory") + else: + path = fsspec_cache_path(identity, cache_dir, ".b2nd") stamp = getattr(src, "stamp", None) if os.path.exists(path) and _cache_stamp(path) != stamp: # The remote frame was replaced, which makes every cached chunk -- and @@ -2015,7 +2049,7 @@ def _open_c2_urlpath(urlpath: blosc2.URLPath, mode: str, offset: int, kwargs: di if offset != 0: raise NotImplementedError("offset is not supported for Caterva2 arrays") - cache_storage = kwargs.pop("cache_storage", None) + cache_dir, cache_path = _remote_cache_options(kwargs) max_concurrency = kwargs.pop("max_concurrency", None) lazy = kwargs.pop("lazy", False) requested = [key for key, value in kwargs.items() if value is not None] @@ -2023,8 +2057,8 @@ def _open_c2_urlpath(urlpath: blosc2.URLPath, mode: str, offset: int, kwargs: di raise NotImplementedError(f"{', '.join(requested)} is not supported for Caterva2 arrays") if not lazy: - if cache_storage is not None: - raise NotImplementedError("cache_storage for a Caterva2 array requires lazy=True") + if cache_dir is not None or cache_path is not None: + raise NotImplementedError("cache_dir and cache_path for a Caterva2 array require lazy=True") if max_concurrency is not None: raise NotImplementedError("max_concurrency is only supported with lazy=True") return blosc2.C2Array(urlpath.path, urlbase=urlpath.urlbase, auth_token=urlpath.auth_token) @@ -2036,7 +2070,7 @@ def _open_c2_urlpath(urlpath: blosc2.URLPath, mode: str, offset: int, kwargs: di # C2Array's constructor has just read api/info. That response supplies both # the geometry and the stamp against which the cache is checked, so asking # for it again in Proxy.__init__ only adds a second serial round trip. - return _lazy_remote_proxy(src, identity, cache_storage, source_fresh=True) + return _lazy_remote_proxy(src, identity, cache_dir, cache_path, source_fresh=True) def _cache_stamp(path: str): @@ -2056,9 +2090,9 @@ def _cache_stamp(path: str): def _open_fsspec_url(urlpath: str, mode: str, offset: int, kwargs: dict): """Open a container living behind an fsspec URL. - Without `cache_storage`, the whole object is fetched in one go and rebuilt in + Without `cache_dir`, the whole object is fetched in one go and rebuilt in memory, which is the right thing for a one-shot read of a small container but - only works for single-file ones. With `cache_storage`, the container is + only works for single-file ones. With `cache_dir`, the container is materialized under that directory and opened as an ordinary local path, so every format, `mmap_mode` and `offset` work. With `lazy`, nothing is fetched up front and each slice pulls just the chunks it needs. @@ -2066,7 +2100,7 @@ def _open_fsspec_url(urlpath: str, mode: str, offset: int, kwargs: dict): if mode != "r": raise NotImplementedError(f"fsspec URLs can only be opened with mode='r', not {mode!r}") - cache_storage = kwargs.pop("cache_storage", None) + cache_dir, cache_path = _remote_cache_options(kwargs) max_concurrency = kwargs.pop("max_concurrency", None) if kwargs.pop("lazy", False): if offset != 0: @@ -2074,25 +2108,28 @@ def _open_fsspec_url(urlpath: str, mode: str, offset: int, kwargs: dict): requested = [k for k, v in kwargs.items() if v is not None] if requested: raise NotImplementedError(f"{', '.join(requested)} is not supported with lazy=True") - return _lazy_fsspec_proxy(urlpath, cache_storage, max_concurrency) + return _lazy_fsspec_proxy(urlpath, cache_dir, cache_path, max_concurrency) + + if cache_path is not None: + raise NotImplementedError("cache_path is only supported with lazy=True") if max_concurrency is not None: # Nothing is fetched chunk by chunk here, so there is nothing to overlap raise NotImplementedError("max_concurrency is only supported with lazy=True") - if cache_storage is not None: - return open(localize_fsspec_url(urlpath, cache_storage), mode, offset, **kwargs) + if cache_dir is not None: + return open(localize_fsspec_url(urlpath, cache_dir), mode, offset, **kwargs) if offset != 0: - raise NotImplementedError("offset on an fsspec URL requires passing cache_storage=") + raise NotImplementedError("offset on an fsspec URL requires passing cache_dir=") # Unset options (dparams=None and friends) are not a request for anything requested = [k for k, v in kwargs.items() if v is not None] if requested: - raise NotImplementedError(f"{', '.join(requested)} on an fsspec URL requires passing cache_storage=") + raise NotImplementedError(f"{', '.join(requested)} on an fsspec URL requires passing cache_dir=") if urlpath.split("?", 1)[0].split("#", 1)[0].endswith(".b2d"): raise NotImplementedError( "directory containers (.b2d, sparse frames) on an fsspec URL require " - "passing cache_storage= to fetch them locally first" + "passing cache_dir= to fetch them locally first" ) with fsspec_open(urlpath, "rb") as f: return blosc2.from_cframe(f.read()) @@ -2124,8 +2161,10 @@ def open( ---------- urlpath: str | pathlib.Path | :ref:`URLPath` The path where the :ref:`SChunk` (or :ref:`NDArray`) - is stored. If it is a remote Caterva2 array, a :ref:`URLPath` must be passed: - a server names its datasets by root and path rather than by URL. + is stored. :ref:`URLPath` is exclusively a Caterva2 dataset reference, + including when its ``urlbase`` is omitted and inherited from + :func:`c2context`; a server names its datasets by root and path rather + than by URL. Any URL with a scheme (``s3://``, ``gs://``, ``https://``, ``zip://``, ``memory://``...) is opened through fsspec; see the `Notes` section for the limits. @@ -2138,7 +2177,7 @@ def open( Open modes also define the allowed persistence side effects: - ``'r'`` never writes to the persistent object. It writes a local cache - only when ``cache_storage`` explicitly requests one; query acceleration + only when ``cache_dir`` or ``cache_path`` explicitly requests one; query acceleration and other implicit execution caches remain process-local only. - ``'a'`` and ``'w'`` may persist explicit user-visible changes such as data, metadata, and index maintenance, but execution caches and query memoization @@ -2148,16 +2187,17 @@ def open( (e.g. in a file containing several such objects). kwargs: dict, optional lazy: bool, optional - For fsspec URLs and Caterva2 :ref:`URLPath` objects, return a - :ref:`Proxy` that leaves the container - where it is and reads what a slice touches, in range requests, - instead of transferring the whole thing. Contiguous frames holding an - :ref:`NDArray` only. A slice landing in a small part of a large chunk - costs only the *blocks* it touches, which for the partitions - :func:`blosc2.asarray` picks by default can be a hundredth of the - chunk; chunks small enough to be one cheap request are still fetched - whole. What arrives is kept in memory, or in ``cache_storage`` when - that is given as well. + For an fsspec URL, return a :ref:`Proxy` over a standalone, + contiguous :ref:`NDArray` frame and read the byte ranges a slice + touches. For a Caterva2 :ref:`URLPath`, return a :ref:`Proxy` over + one array-like dataset; stored ``.b2nd`` arrays use byte ranges when + available, while HDF5 datasets, ``.b2z`` leaves, and computed arrays + fall back to semantic chunk requests. Neither form opens a whole + remote store hierarchy. A slice landing in a small part of a large + chunk costs only the *blocks* it touches when ranges are available; + chunks small enough to be one cheap request are still fetched whole. + What arrives is kept in memory, under ``cache_dir``, or at the exact + ``cache_path`` when either is given. max_concurrency: int, optional Only with ``lazy``: how many fetches to run at once, in a thread pool. A slice against an object store is almost entirely round-trip @@ -2165,14 +2205,19 @@ def open( bearable. Defaults to 8; pass 1 for a protocol with no latency to hide, where the pool costs about 10 microseconds per chunk and saves nothing. - cache_storage: str | pathlib.Path, optional - For fsspec URLs and lazy Caterva2 :ref:`URLPath` objects, a directory - holding this container's local + cache_dir: str | pathlib.Path, optional + For fsspec URLs and lazy Caterva2 :ref:`URLPath` objects, a directory holding this container's local copy — the whole thing, or just the chunks and blocks ``lazy`` has fetched so far. Either way a later run starts from what is already there, and the copy is discarded when the remote no longer matches it. There is no default on purpose, so nothing writes to a disk you did not name. + cache_path: str | pathlib.Path, optional + With ``lazy=True``, the exact file to use for the remote array's + persistent proxy cache. Mutually exclusive with ``cache_dir``. + cache_storage: str | pathlib.Path, optional + Deprecated alias for ``cache_dir``. Mutually exclusive with + ``cache_dir`` and ``cache_path``. mmap_mode: str, optional If set, the file will be memory-mapped instead of using the default I/O functions and the `mode` argument will be ignored. @@ -2214,8 +2259,8 @@ def open( * If :paramref:`urlpath` is a :ref:`URLPath` instance, :paramref:`mode` must be 'r' and :paramref:`offset` must be 0. Without ``lazy=True`` it returns a :ref:`C2Array`; with ``lazy=True`` it returns a :ref:`Proxy`, - optionally persisted under ``cache_storage``. Authenticated users sharing - a machine must use separate cache directories. + optionally persisted under ``cache_dir`` or at ``cache_path``. + Authenticated users sharing a machine must use separate caches. * fsspec URLs need the ``fsspec`` extra (``pip install "blosc2[fsspec]"``) and the driver for the protocol (``s3fs``, ``gcsfs``...), which fsspec asks for @@ -2223,8 +2268,8 @@ def open( ``mode != 'r'`` always raises, as object stores have no rename and no locks. A plain URL read rebuilds the object from a cframe held in memory, so it covers ``.b2nd``, ``.b2f`` and ``.b2e`` only -- a ``.b2z`` store is a zip - archive rather than a cframe, and needs ``cache_storage`` like the - directory formats do. ``cache_storage`` and ``lazy`` above lift that, each + archive rather than a cframe, and needs ``cache_dir`` like the directory + formats do. ``cache_dir`` and ``lazy`` above lift that, each in its own way. * Persistent data handling follows a strict no-hidden-writes rule: diff --git a/tests/ndarray/test_c2array_blocks.py b/tests/ndarray/test_c2array_blocks.py index 0c57d3c7e..50eeb3c64 100644 --- a/tests/ndarray/test_c2array_blocks.py +++ b/tests/ndarray/test_c2array_blocks.py @@ -349,20 +349,37 @@ def test_open_urlpath_lazy_persistent_cache(tmp_path, server, any_chunk_wants_bl data = _incompressible((200, 200)) array, srv = server(data, chunks=(100, 200), blocks=(10, 20)) urlpath = blosc2.URLPath(array.path, urlbase=array.urlbase) - cache_storage = tmp_path / "cache" + cache_dir = tmp_path / "cache" srv.log.clear() - proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + proxy = blosc2.open(urlpath, lazy=True, cache_dir=cache_dir) assert [endpoint for endpoint, _, _ in srv.log] == ["info"] assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) del proxy srv.log.clear() - proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + proxy = blosc2.open(urlpath, lazy=True, cache_dir=cache_dir) assert [endpoint for endpoint, _, _ in srv.log] == ["info"] assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) assert [endpoint for endpoint, _, _ in srv.log] == ["info"] - assert len(list(cache_storage.glob("*.b2nd"))) == 1 + assert len(list(cache_dir.glob("*.b2nd"))) == 1 + + +def test_open_urlpath_lazy_exact_cache_path(tmp_path, server, any_chunk_wants_blocks): + data = _incompressible((200, 200)) + array, srv = server(data, chunks=(100, 200), blocks=(10, 20)) + urlpath = blosc2.URLPath(array.path, urlbase=array.urlbase) + cache_path = tmp_path / "chosen.b2nd" + + proxy = blosc2.open(urlpath, lazy=True, cache_path=cache_path) + assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) + assert proxy.urlpath == str(cache_path) + del proxy + + srv.log.clear() + proxy = blosc2.open(urlpath, lazy=True, cache_path=cache_path) + assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) + assert [endpoint for endpoint, _, _ in srv.log] == ["info"] def test_open_urlpath_lazy_uses_c2context_without_persisting_token(tmp_path, server): @@ -370,14 +387,14 @@ def test_open_urlpath_lazy_uses_c2context_without_persisting_token(tmp_path, ser data = _incompressible((20, 20)) array, _ = server(data, chunks=(10, 20), blocks=(5, 10), cookie=token) urlpath = blosc2.URLPath(array.path) - cache_storage = tmp_path / "cache" + cache_dir = tmp_path / "cache" with blosc2.c2context(urlbase=array.urlbase, auth_token=token): - proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + proxy = blosc2.open(urlpath, lazy=True, cache_dir=cache_dir) assert np.array_equal(proxy[0:5, 0:5], data[0:5, 0:5]) assert proxy.schunk.meta["proxy-source"]["urlpath"][2] is None - cache = next(cache_storage.glob("*.b2nd")) + cache = next(cache_dir.glob("*.b2nd")) reopened = blosc2.open(cache, mode="a") assert np.array_equal(reopened[0:5, 0:5], data[0:5, 0:5]) @@ -386,16 +403,16 @@ def test_open_urlpath_lazy_rebuilds_stale_cache(tmp_path, server, any_chunk_want data = _incompressible((200, 200)) array, srv = server(data, chunks=(100, 200), blocks=(10, 20)) urlpath = blosc2.URLPath(array.path, urlbase=array.urlbase) - cache_storage = tmp_path / "cache" + cache_dir = tmp_path / "cache" - proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + proxy = blosc2.open(urlpath, lazy=True, cache_dir=cache_dir) assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) del proxy other = _incompressible((200, 200), seed=1) _replace(srv, other, chunks=(100, 200), blocks=(10, 20)) - proxy = blosc2.open(urlpath, lazy=True, cache_storage=cache_storage) + proxy = blosc2.open(urlpath, lazy=True, cache_dir=cache_dir) assert np.array_equal(proxy[0:5, 0:10], other[0:5, 0:10]) @@ -405,8 +422,8 @@ def test_open_urlpath_cache_options_need_lazy(tmp_path, server): urlpath = blosc2.URLPath(array.path, urlbase=array.urlbase) assert isinstance(blosc2.open(urlpath), blosc2.C2Array) - with pytest.raises(NotImplementedError, match=r"cache_storage.*lazy=True"): - blosc2.open(urlpath, cache_storage=tmp_path) + with pytest.raises(NotImplementedError, match=r"cache_dir.*lazy=True"): + blosc2.open(urlpath, cache_dir=tmp_path) with pytest.raises(NotImplementedError, match=r"max_concurrency.*lazy=True"): blosc2.open(urlpath, max_concurrency=2) diff --git a/tests/test_fsspec.py b/tests/test_fsspec.py index cd0499f8e..93f138a35 100644 --- a/tests/test_fsspec.py +++ b/tests/test_fsspec.py @@ -122,26 +122,26 @@ def test_chained_url(tmp_path): @pytest.mark.parametrize("mode", ["a", "w"]) def test_mode_not_supported(mode): with pytest.raises(NotImplementedError): - blosc2.open("memory://x.b2nd", mode=mode, cache_storage="/tmp/nope") + blosc2.open("memory://x.b2nd", mode=mode, cache_dir="/tmp/nope") def test_offset_needs_cache(): - with pytest.raises(NotImplementedError, match="cache_storage"): + with pytest.raises(NotImplementedError, match="cache_dir"): blosc2.open("memory://x.b2nd", offset=32) def test_mmap_needs_cache(): - with pytest.raises(NotImplementedError, match="cache_storage"): + with pytest.raises(NotImplementedError, match="cache_dir"): blosc2.open("memory://x.b2nd", mmap_mode="r") def test_dir_container_needs_cache(): - with pytest.raises(NotImplementedError, match="cache_storage"): + with pytest.raises(NotImplementedError, match="cache_dir"): blosc2.open("memory://store.b2d") def test_dir_container_with_query_needs_cache(): - with pytest.raises(NotImplementedError, match="cache_storage"): + with pytest.raises(NotImplementedError, match="cache_dir"): blosc2.open("memory://store.b2d?version=1") @@ -150,7 +150,7 @@ def test_cached_open(tmp_path): with fsspec.open("memory://c.b2nd", "wb") as f: f.write(a.to_cframe()) - b = blosc2.open("memory://c.b2nd", cache_storage=tmp_path) + b = blosc2.open("memory://c.b2nd", cache_dir=tmp_path) assert np.array_equal(b[:], a[:]) assert any(tmp_path.iterdir()) @@ -161,7 +161,7 @@ def test_cached_open_is_local(tmp_path): with fsspec.open("memory://m.b2nd", "wb") as f: f.write(a.to_cframe()) - b = blosc2.open("memory://m.b2nd", cache_storage=tmp_path, mmap_mode="r") + b = blosc2.open("memory://m.b2nd", cache_dir=tmp_path, mmap_mode="r") assert np.array_equal(b[:], a[:]) @@ -177,20 +177,20 @@ def test_cache_hit_avoids_refetch(tmp_path, monkeypatch): memfs, "_open", lambda self, path, *a, **kw: (fetches.append(path), orig(self, path, *a, **kw))[1] ) - blosc2.open("memory://h.b2nd", cache_storage=tmp_path) + blosc2.open("memory://h.b2nd", cache_dir=tmp_path) assert len(fetches) == 1 - blosc2.open("memory://h.b2nd", cache_storage=tmp_path) + blosc2.open("memory://h.b2nd", cache_dir=tmp_path) assert len(fetches) == 1 def test_cache_refetches_when_remote_changes(tmp_path): with fsspec.open("memory://s.b2nd", "wb") as f: f.write(blosc2.arange(10, dtype="i4").to_cframe()) - assert blosc2.open("memory://s.b2nd", cache_storage=tmp_path).shape == (10,) + assert blosc2.open("memory://s.b2nd", cache_dir=tmp_path).shape == (10,) with fsspec.open("memory://s.b2nd", "wb") as f: f.write(blosc2.arange(20, dtype="i4").to_cframe()) - assert blosc2.open("memory://s.b2nd", cache_storage=tmp_path).shape == (20,) + assert blosc2.open("memory://s.b2nd", cache_dir=tmp_path).shape == (20,) def test_cached_dict_store(tmp_path): @@ -201,7 +201,7 @@ def test_cached_dict_store(tmp_path): dstore["/b"] = blosc2.arange(5, dtype="f8") fsspec.filesystem("memory").put(localstore, "memory://store.b2d", recursive=True) - with blosc2.open("memory://store.b2d", cache_storage=tmp_path / "cache") as dstore: + with blosc2.open("memory://store.b2d", cache_dir=tmp_path / "cache") as dstore: assert sorted(dstore.keys()) == ["/a", "/b"] assert np.array_equal(dstore["/a"][:], np.arange(10, dtype="i4")) @@ -213,14 +213,14 @@ def test_cached_dir_refetches_when_remote_changes(tmp_path): with blosc2.DictStore(localstore, mode="w") as dstore: dstore["/a"] = blosc2.arange(10, dtype="i4") memfs.put(localstore, "memory://d.b2d", recursive=True) - with blosc2.open("memory://d.b2d", cache_storage=cache) as dstore: + with blosc2.open("memory://d.b2d", cache_dir=cache) as dstore: assert list(dstore.keys()) == ["/a"] with blosc2.DictStore(localstore, mode="a") as dstore: dstore["/b"] = blosc2.arange(5, dtype="i4") memfs.rm("/d.b2d", recursive=True) memfs.put(localstore, "memory://d.b2d", recursive=True) - with blosc2.open("memory://d.b2d", cache_storage=cache) as dstore: + with blosc2.open("memory://d.b2d", cache_dir=cache) as dstore: assert sorted(dstore.keys()) == ["/a", "/b"] @@ -229,7 +229,7 @@ def test_cached_sparse_frame(tmp_path): a = blosc2.arange(1000, dtype="i4", chunks=(100,), urlpath=localpath, mode="w", contiguous=False) fsspec.filesystem("memory").put(localpath, "memory://sparse.b2nd", recursive=True) - b = blosc2.open("memory://sparse.b2nd", cache_storage=tmp_path / "cache") + b = blosc2.open("memory://sparse.b2nd", cache_dir=tmp_path / "cache") assert np.array_equal(b[:], a[:]) @@ -470,7 +470,7 @@ def test_lazy_rejects_directories(tmp_path): localpath = str(tmp_path / "sparse.b2nd") blosc2.arange(0, 1000, dtype="i4", chunks=(100,), urlpath=localpath, mode="w", contiguous=False) fsspec.filesystem("memory").put(localpath, "memory://sparse.b2nd", recursive=True) - with pytest.raises(NotImplementedError, match="cache_storage"): + with pytest.raises(NotImplementedError, match="cache_dir"): blosc2.open("memory://sparse.b2nd", lazy=True) @@ -480,7 +480,7 @@ def test_lazy_not_a_frame(): blosc2.open("memory://junk.b2nd", lazy=True) -def test_lazy_with_cache_storage(tmp_path, monkeypatch): +def test_lazy_with_cache_dir(tmp_path, monkeypatch): a = blosc2.arange(0, 1000, dtype="i4", chunks=(100,)) url = _put("lazycache.b2nd", a) @@ -492,19 +492,46 @@ def test_lazy_with_cache_storage(tmp_path, monkeypatch): lambda self, nchunk: (fetched.append(nchunk), orig(self, nchunk))[1], ) - p = blosc2.open(url, lazy=True, cache_storage=tmp_path) + p = blosc2.open(url, lazy=True, cache_dir=tmp_path) assert np.array_equal(p[0:100], a[0:100]) assert fetched == [0] del p # A later run starts from the chunks the previous one pulled - p = blosc2.open(url, lazy=True, cache_storage=tmp_path) + p = blosc2.open(url, lazy=True, cache_dir=tmp_path) assert np.array_equal(p[0:100], a[0:100]) assert fetched == [0] assert np.array_equal(p[500:600], a[500:600]) assert fetched == [0, 5] +def test_lazy_with_exact_cache_path(tmp_path): + a = blosc2.arange(0, 1000, dtype="i4", chunks=(100,)) + url = _put("exactcache.b2nd", a) + cache_path = tmp_path / "chosen.b2nd" + + p = blosc2.open(url, lazy=True, cache_path=cache_path) + assert np.array_equal(p[0:100], a[0:100]) + assert p.urlpath == str(cache_path) + assert cache_path.is_file() + + q = blosc2.open(url, lazy=True, cache_path=cache_path) + assert np.array_equal(q[0:100], a[0:100]) + + +def test_remote_cache_options_are_mutually_exclusive(tmp_path): + url = _put("exclusivecache.b2nd", blosc2.arange(0, 10)) + with pytest.raises(ValueError, match="mutually exclusive"): + blosc2.open(url, lazy=True, cache_dir=tmp_path, cache_path=tmp_path / "cache.b2nd") + + +def test_cache_storage_is_deprecated(tmp_path): + url = _put("legacycache.b2nd", blosc2.arange(0, 10)) + with pytest.warns(DeprecationWarning, match="use cache_dir"): + p = blosc2.open(url, lazy=True, cache_storage=tmp_path) + assert np.array_equal(p[:], np.arange(10)) + + def test_lazy_cache_rebuilt_when_remote_changes(tmp_path): # Uncompressed, so both frames are byte-for-byte the same size: the stamp # cannot fall back to comparing sizes and get this right by luck @@ -514,14 +541,14 @@ def test_lazy_cache_rebuilt_when_remote_changes(tmp_path): assert len(a.to_cframe()) == len(b.to_cframe()) url = _put("lazystale.b2nd", a) - p = blosc2.open(url, lazy=True, cache_storage=tmp_path) + p = blosc2.open(url, lazy=True, cache_dir=tmp_path) assert np.array_equal(p[0:100], a[0:100]) del p # Replacing the frame invalidates both the cached chunks and the offsets # they were fetched by, so the cache must be thrown away rather than reused _put("lazystale.b2nd", b) - p = blosc2.open(url, lazy=True, cache_storage=tmp_path) + p = blosc2.open(url, lazy=True, cache_dir=tmp_path) assert np.array_equal(p[0:100], b[0:100]) @@ -554,7 +581,7 @@ def test_http_url_is_read_through_fsspec(tmp_path): assert np.array_equal(whole[:], data) requests.clear() - lazy = blosc2.open(f"{urlbase}/big.b2nd", lazy=True, cache_storage=str(tmp_path / "cs")) + lazy = blosc2.open(f"{urlbase}/big.b2nd", lazy=True, cache_dir=str(tmp_path / "cs")) assert isinstance(lazy, blosc2.Proxy) assert isinstance(lazy.src, blosc2.FsspecNDSource) assert ":etag:" in lazy.src.stamp @@ -574,12 +601,12 @@ def test_http_lazy_cache_rebuilt_when_remote_changes(tmp_path): with _ranged_server(path) as (urlbase, _): url = f"{urlbase}/{frame.name}" cache = tmp_path / "cache" - lazy = blosc2.open(url, lazy=True, cache_storage=cache) + lazy = blosc2.open(url, lazy=True, cache_dir=cache) assert np.array_equal(lazy[3:5, 100:120], first[3:5, 100:120]) del lazy blosc2.asarray(second, chunks=(50, 200), blocks=(10, 100), urlpath=frame, mode="w") - lazy = blosc2.open(url, lazy=True, cache_storage=cache) + lazy = blosc2.open(url, lazy=True, cache_dir=cache) assert np.array_equal(lazy[3:5, 100:120], second[3:5, 100:120]) @@ -645,7 +672,7 @@ def test_zip_store_needs_cache(tmp_path): with pytest.raises(RuntimeError): blosc2.open("memory://t.b2z") - with blosc2.open("memory://t.b2z", cache_storage=tmp_path / "cache") as tstore: + with blosc2.open("memory://t.b2z", cache_dir=tmp_path / "cache") as tstore: assert np.array_equal(tstore["/a"][:], np.arange(10, dtype="i4")) @@ -729,7 +756,7 @@ def test_cached_container_keeps_its_extension(tmp_path): del estore fsspec.filesystem("memory").pipe_file("/e.b2e", pathlib.Path(localpath).read_bytes()) - opened = blosc2.open("memory://e.b2e", cache_storage=tmp_path / "cache") + opened = blosc2.open("memory://e.b2e", cache_dir=tmp_path / "cache") assert isinstance(opened, blosc2.EmbedStore) assert np.array_equal(opened["/a"][:], np.arange(10, dtype="i4")) @@ -747,23 +774,23 @@ def test_lazy_empty_array(tmp_path): def test_lazy_cache_rebuilt_when_corrupt(tmp_path): # An interrupted run can leave a half-written cache behind; the whole point of - # cache_storage is surviving across runs, so it has to be discarded, not fatal + # cache_dir is surviving across runs, so it has to be discarded, not fatal a = blosc2.arange(100, dtype="i4", chunks=(10,)) fsspec.filesystem("memory").pipe_file("/c.b2nd", a.to_cframe()) - with blosc2.open("memory://c.b2nd", lazy=True, cache_storage=tmp_path) as b: + with blosc2.open("memory://c.b2nd", lazy=True, cache_dir=tmp_path) as b: assert np.array_equal(b[:10], a[:10]) cache = next(p for p in tmp_path.iterdir() if p.suffix == ".b2nd") cache.write_bytes(cache.read_bytes()[:50]) - with blosc2.open("memory://c.b2nd", lazy=True, cache_storage=tmp_path) as b: + with blosc2.open("memory://c.b2nd", lazy=True, cache_dir=tmp_path) as b: assert np.array_equal(b[:], a[:]) def test_max_concurrency_needs_lazy(tmp_path): fsspec.filesystem("memory").pipe_file("/m.b2nd", blosc2.arange(10, dtype="i4").to_cframe()) with pytest.raises(NotImplementedError, match="max_concurrency"): - blosc2.open("memory://m.b2nd", cache_storage=tmp_path, max_concurrency=4) + blosc2.open("memory://m.b2nd", cache_dir=tmp_path, max_concurrency=4) def test_storage_mapping_is_normalized(tmp_path): @@ -1337,13 +1364,13 @@ def test_lazy_eviction_survives_a_reopen(tmp_path, monkeypatch, any_chunk_wants_ cache = str(tmp_path / "evicted-cache") reads, _ = _traffic(monkeypatch) - p = blosc2.open(url, lazy=True, cache_storage=cache) + p = blosc2.open(url, lazy=True, cache_dir=cache) assert np.array_equal(p[0:5, 0:10], data[0:5, 0:10]) fetched = len(reads) p.schunk.update_special(0, blosc2.SpecialValue.UNINIT) del p - q = blosc2.open(url, lazy=True, cache_storage=cache) + q = blosc2.open(url, lazy=True, cache_dir=cache) assert np.array_equal(q[0:5, 0:10], data[0:5, 0:10]) assert len(reads) > fetched diff --git a/tests/test_fsspec_s3.py b/tests/test_fsspec_s3.py index f4f8d583a..61b148d54 100644 --- a/tests/test_fsspec_s3.py +++ b/tests/test_fsspec_s3.py @@ -81,9 +81,9 @@ def test_save_and_open_whole(stored): assert np.array_equal(blosc2.open(urlpath)[:], a[:]) -def test_cache_storage(stored, tmp_path): +def test_cache_dir(stored, tmp_path): urlpath, a = stored - b = blosc2.open(urlpath, cache_storage=tmp_path, mmap_mode="r") + b = blosc2.open(urlpath, cache_dir=tmp_path, mmap_mode="r") assert np.array_equal(b[:], a[:]) From eb214c197a65a4d482393a1db5640e079b92b46c Mon Sep 17 00:00:00 2001 From: Francesc Alted Date: Fri, 4 Sep 2026 08:03:25 +0200 Subject: [PATCH 06/10] Reconstruct remote sources from proxy caches --- doc/guides/remote_arrays.md | 18 +++++++++++++++++ src/blosc2/proxy.py | 17 +++++++++++++--- src/blosc2/schunk.py | 29 +++++++++++++++++++++------- tests/ndarray/test_c2array_blocks.py | 7 ++++++- tests/test_fsspec.py | 28 +++++++++++++++++++++++++++ 5 files changed, 88 insertions(+), 11 deletions(-) diff --git a/doc/guides/remote_arrays.md b/doc/guides/remote_arrays.md index 4010b67fa..4da6247fc 100644 --- a/doc/guides/remote_arrays.md +++ b/doc/guides/remote_arrays.md @@ -83,6 +83,24 @@ In both cases, the cache is an ordinary `.b2nd` array that starts small and grow Authenticated Caterva2 caches must be private to one user. Reopen them under an equivalent authenticated {func}`blosc2.c2context`; do not share a cache directory between users. +### Reopen a cache file independently + +A persistent cache records enough information to reconstruct built-in fsspec and Caterva2 sources. When its filename is known, it can therefore be opened without repeating the original remote URL: + +```python +# Created earlier with cache_path="big-cache.b2nd" +a = blosc2.open("big-cache.b2nd", mode="a") + +a[100:110, :50] # cached data stays local +a[500:510, :50] # missing data is fetched from the recorded source and cached +``` + +This cache is operational, but not necessarily self-contained. Regions not fetched previously still require the original source. Opening with `mode="a"` lets newly fetched regions extend the cache; the default `mode="r"` keeps the cache file unchanged. + +Independent reopening works for fsspec URLs, Caterva2 datasets, and persistent local Blosc2 sources. The required runtime environment must still be available: fsspec backends and their configuration must be installed, local source paths must remain valid, and authenticated Caterva2 caches must be reopened inside an equivalent {func}`blosc2.c2context`. Caterva2 credentials are not stored in the cache file. + +An arbitrary custom {ref}`ProxyNDSource` cannot be reconstructed because its Python class and runtime state are not serialized. In that case, recreate the source explicitly and attach the existing cache with `blosc2.Proxy(source, urlpath="big-cache.b2nd", mode="a")`. + ## Only what a slice touches Blosc2 arrays are compressed in chunks, which are divided into smaller blocks. For a small slice, fetching only its blocks can avoid transferring most of a large chunk. diff --git a/src/blosc2/proxy.py b/src/blosc2/proxy.py index 7d76d3378..c2b1160e7 100644 --- a/src/blosc2/proxy.py +++ b/src/blosc2/proxy.py @@ -172,15 +172,26 @@ def __init__( fresh = self._cache is None if fresh: meta_val = { + "source_kind": None, "local_abspath": None, "urlpath": None, "caterva2_env": caterva2_env, } container = getattr(self.src, "schunk", self.src) - if hasattr(container, "urlpath"): - meta_val["local_abspath"] = container.urlpath + if isinstance(self.src, blosc2.FsspecNDSource): + meta_val["source_kind"] = "fsspec" + meta_val["urlpath"] = self.src.urlpath + # Keep the legacy field populated so older readers still + # reopen this cache, albeit through their eager URL path. + meta_val["local_abspath"] = self.src.urlpath elif isinstance(self.src, blosc2.C2Array): - meta_val["urlpath"] = (self.src.path, self.src.urlbase, self.src.auth_token) + meta_val["source_kind"] = "caterva2" + # Authentication belongs to the reopening process, not to a + # portable cache file. C2Array resolves it again from c2context. + meta_val["urlpath"] = (self.src.path, self.src.urlbase, None) + elif hasattr(container, "urlpath"): + meta_val["source_kind"] = "local" + meta_val["local_abspath"] = container.urlpath meta = {"proxy-source": meta_val} if hasattr(self.src, "shape"): self._cache = blosc2.empty( diff --git a/src/blosc2/schunk.py b/src/blosc2/schunk.py index ac23c6a16..7fe4925d7 100644 --- a/src/blosc2/schunk.py +++ b/src/blosc2/schunk.py @@ -1891,9 +1891,25 @@ def process_opened_object(res): if "proxy-source" in meta: proxy_cache = res proxy_src = meta["proxy-source"] + source_kind = proxy_src.get("source_kind") + if source_kind == "fsspec": + src = blosc2.FsspecNDSource(proxy_src["urlpath"]) + return blosc2.Proxy(src, _cache=proxy_cache, _refresh_source=False) + if source_kind == "caterva2": + src = blosc2.C2Array( + proxy_src["urlpath"][0], proxy_src["urlpath"][1], proxy_src["urlpath"][2] + ) + return blosc2.Proxy(src, _cache=proxy_cache, _refresh_source=False) if proxy_src["local_abspath"] is not None: - src = blosc2.open(proxy_src["local_abspath"], mode="r") - return blosc2.Proxy(src, _cache=proxy_cache) + source_path = proxy_src["local_abspath"] + # Older FsspecNDSource caches recorded their URL in the field that + # otherwise names a local source. Preserve those caches while + # restoring their lazy byte-range behavior. + if source_kind is None and is_fsspec_url(source_path): + src = blosc2.FsspecNDSource(source_path) + else: + src = blosc2.open(source_path, mode="r") + return blosc2.Proxy(src, _cache=proxy_cache, _refresh_source=False) elif proxy_src["urlpath"] is not None: src = blosc2.C2Array(proxy_src["urlpath"][0], proxy_src["urlpath"][1], proxy_src["urlpath"][2]) return blosc2.Proxy(src, _cache=proxy_cache) @@ -2279,11 +2295,10 @@ def open( caller; runtime caches are not serialized back to disk. * If the original object saved in :paramref:`urlpath` is a :ref:`Proxy`, - this function will only return a :ref:`Proxy` if its source is a local - :ref:`SChunk`, :ref:`NDArray` or a remote :ref:`C2Array`. Otherwise, - it will return the Python-Blosc2 container used to cache the data which - can be a :ref:`SChunk` or a :ref:`NDArray` and may not have all the data - initialized (e.g. if the user has not accessed to it yet). + this function reconstructs sources backed by a persistent local + :ref:`SChunk` or :ref:`NDArray`, an fsspec URL, or a remote + :ref:`C2Array`. Custom proxy sources must be recreated explicitly because + their Python class and runtime state are not stored in the cache. * When opening a :ref:`LazyExpr` keep in mind the note above regarding operands. diff --git a/tests/ndarray/test_c2array_blocks.py b/tests/ndarray/test_c2array_blocks.py index 50eeb3c64..cf11c85a5 100644 --- a/tests/ndarray/test_c2array_blocks.py +++ b/tests/ndarray/test_c2array_blocks.py @@ -374,12 +374,17 @@ def test_open_urlpath_lazy_exact_cache_path(tmp_path, server, any_chunk_wants_bl proxy = blosc2.open(urlpath, lazy=True, cache_path=cache_path) assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) assert proxy.urlpath == str(cache_path) + assert proxy.schunk.meta["proxy-source"]["source_kind"] == "caterva2" del proxy srv.log.clear() - proxy = blosc2.open(urlpath, lazy=True, cache_path=cache_path) + proxy = blosc2.open(cache_path, mode="a") + assert isinstance(proxy, blosc2.Proxy) + assert isinstance(proxy.src, blosc2.C2Array) assert np.array_equal(proxy[0:5, 0:10], data[0:5, 0:10]) assert [endpoint for endpoint, _, _ in srv.log] == ["info"] + assert np.array_equal(proxy[100:105, 0:10], data[100:105, 0:10]) + assert len(srv.log) > 1 def test_open_urlpath_lazy_uses_c2context_without_persisting_token(tmp_path, server): diff --git a/tests/test_fsspec.py b/tests/test_fsspec.py index 93f138a35..29d41d8d5 100644 --- a/tests/test_fsspec.py +++ b/tests/test_fsspec.py @@ -519,6 +519,34 @@ def test_lazy_with_exact_cache_path(tmp_path): assert np.array_equal(q[0:100], a[0:100]) +def test_exact_cache_path_reopens_as_lazy_fsspec_proxy(tmp_path, monkeypatch): + a = blosc2.arange(0, 1000, dtype="i4", chunks=(100,)) + url = _put("independentcache.b2nd", a) + cache_path = tmp_path / "independent.b2nd" + p = blosc2.open(url, lazy=True, cache_path=cache_path) + assert np.array_equal(p[0:100], a[0:100]) + source_meta = p.schunk.meta["proxy-source"] + assert source_meta["source_kind"] == "fsspec" + assert source_meta["urlpath"] == url + del p + + fetched = [] + orig = blosc2.FsspecNDSource.get_chunk + monkeypatch.setattr( + blosc2.FsspecNDSource, + "get_chunk", + lambda self, nchunk: (fetched.append(nchunk), orig(self, nchunk))[1], + ) + + reopened = blosc2.open(cache_path, mode="a") + assert isinstance(reopened, blosc2.Proxy) + assert isinstance(reopened.src, blosc2.FsspecNDSource) + assert np.array_equal(reopened[0:100], a[0:100]) + assert fetched == [] + assert np.array_equal(reopened[500:600], a[500:600]) + assert fetched == [5] + + def test_remote_cache_options_are_mutually_exclusive(tmp_path): url = _put("exclusivecache.b2nd", blosc2.arange(0, 10)) with pytest.raises(ValueError, match="mutually exclusive"): From 266e8e8fc3c45a83e3445ea81b4845ddacf70361 Mon Sep 17 00:00:00 2001 From: Francesc Alted Date: Fri, 4 Sep 2026 08:16:25 +0200 Subject: [PATCH 07/10] Count metadata in remote traffic --- doc/guides/remote_arrays.md | 2 +- examples/c2array-traffic.py | 7 +++---- examples/fsspec-cat2-access.py | 15 +++++++++------ src/blosc2/c2array.py | 25 ++++++++++++++++++------- src/blosc2/proxy.py | 7 +++---- tests/ndarray/test_c2array_blocks.py | 6 ++++-- 6 files changed, 38 insertions(+), 24 deletions(-) diff --git a/doc/guides/remote_arrays.md b/doc/guides/remote_arrays.md index 4da6247fc..5aec9e9ae 100644 --- a/doc/guides/remote_arrays.md +++ b/doc/guides/remote_arrays.md @@ -115,7 +115,7 @@ Stepped slices also use the block grid. For example, `p[::5]` can reduce transfe ### Measure network traffic -{ref}`C2Array` and remote {ref}`Proxy` objects expose cumulative request and byte counts through {ref}`Traffic`: +{ref}`C2Array` and remote {ref}`Proxy` objects expose cumulative request and byte counts through {ref}`Traffic`. The count starts when the remote source is opened, so it includes metadata as well as array data: ```python source = blosc2.C2Array( diff --git a/examples/c2array-traffic.py b/examples/c2array-traffic.py index 60b323b76..ee8b42835 100644 --- a/examples/c2array-traffic.py +++ b/examples/c2array-traffic.py @@ -11,7 +11,7 @@ # the whole chunk: on a fast link the two take about as long, and differ by the # compression ratio in bytes. Bytes are also what a metered link and a shared # server uplink actually run out of, so they are what `Traffic` counts -- at the -# transport, so the frame index and block offsets are in the tally too. +# transport, so metadata, the frame index, and block offsets are in the tally too. import blosc2 @@ -27,9 +27,8 @@ def cost(traffic): array = blosc2.C2Array(path, urlbase=urlbase) print(f"{path}: shape={array.shape} chunks={array.chunks} blocks={array.blocks}") -# Opening a handle costs one `api/info` call, which is metadata rather than data -# and is deliberately not counted -- no slice can avoid it, and no choice of -# granularity changes it. +# Opening a handle costs one `api/info` call, included so this is a complete +# account of everything that crossed the wire. print(f"after opening: {array.traffic}") # -- A proxy reads through the block path, so it pays for what a slice touches. diff --git a/examples/fsspec-cat2-access.py b/examples/fsspec-cat2-access.py index 78e21cfbf..4f685a0fd 100644 --- a/examples/fsspec-cat2-access.py +++ b/examples/fsspec-cat2-access.py @@ -39,7 +39,8 @@ def traffic_text(traffic: blosc2.Traffic | None) -> str: if traffic is None: return "traffic unavailable" - return f"{traffic.requests} requests, {traffic.nbytes / 2**20:.3f} MiB" + request_word = "request" if traffic.requests == 1 else "requests" + return f"{traffic.requests} {request_word}, {traffic.nbytes / 2**20:.3f} MiB" def size_text(size: int) -> str: @@ -52,6 +53,7 @@ def benchmark(label: str, urlpath, cache_dir: Path) -> np.ndarray: start = perf_counter() array = blosc2.open(urlpath, lazy=True, cache_dir=cache_dir) open_time = perf_counter() - start + open_traffic = traffic_text(array.traffic) metadata = (array.shape, array.dtype, array.chunks, array.blocks) cache_path = Path(array.urlpath).resolve() @@ -69,6 +71,7 @@ def benchmark(label: str, urlpath, cache_dir: Path) -> np.ndarray: start = perf_counter() reopened = blosc2.open(urlpath, lazy=True, cache_dir=cache_dir) reopen_time = perf_counter() - start + reopen_traffic = traffic_text(reopened.traffic) reopened.traffic.reset() start = perf_counter() @@ -81,11 +84,11 @@ def benchmark(label: str, urlpath, cache_dir: Path) -> np.ndarray: print(f" metadata: shape={metadata[0]}, dtype={metadata[1]}") print(f" chunks={metadata[2]}, blocks={metadata[3]}") print(f" persistent cache: {cache_path} ({'existing' if cache_existed else 'new'})") - print(f" open and remote metadata setup: {open_time:.6f} s") - print(f" first data slice this run: {first_read_time:.6f} s ({first_traffic})") - print(f" cache size after slice: {size_text(cache_size)}") - print(f" reopen persistent cache: {reopen_time:.6f} s") - print(f" same slice after reopen: {cached_read_time:.6f} s ({cached_traffic})") + print(f" {'open + remote metadata:':<27}{open_time * 1000:.0f} ms ({open_traffic})") + print(f" {'first data slice:':<27}{first_read_time * 1000:.0f} ms ({first_traffic})") + print(f" {'cache after slice:':<27}{size_text(cache_size)}") + print(f" {'reopen + remote metadata:':<27}{reopen_time * 1000:.0f} ms ({reopen_traffic})") + print(f" {'same slice after reopen:':<27}{cached_read_time * 1000:.0f} ms ({cached_traffic})") return data diff --git a/src/blosc2/c2array.py b/src/blosc2/c2array.py index 9286894b9..c14db8303 100644 --- a/src/blosc2/c2array.py +++ b/src/blosc2/c2array.py @@ -264,9 +264,11 @@ def login(username, password, urlbase): return "=".join(list(resp.cookies.items())[0]) -def info(path, urlbase, params=None, headers=None, model=None, auth_token=None): +def info(path, urlbase, params=None, headers=None, model=None, auth_token=None, traffic=None): url = _server_url(urlbase, f"api/info/{path}") response = _xget(url, params, headers, auth_token) + if traffic is not None: + traffic.charge(len(response.content)) json = response.json() return json if model is None else model(**json) @@ -771,10 +773,9 @@ def __init__(self, path: str, /, urlbase: str | None = None, auth_token: str | N """Bytes and requests this handle has read off the server; see :ref:`Traffic`. Cumulative since the array was opened, counted at the transport, so the - frame index and the block offsets are in it as well as the data, and the - `api/info` call that opened this handle is not. Whichever endpoint the - read used is in it too, and the block source built later is handed this - same tally, so one counter answers for the array however it is read. + opening `api/info` response, frame index, block offsets, and data are all + included. Whichever endpoint serves a read uses this same tally, so one + counter answers for the array however it is read. What a slice cost is the difference between two readings, or one reading after :meth:`Traffic.reset`. `examples/c2array-traffic.py` is a runnable @@ -784,7 +785,12 @@ def __init__(self, path: str, /, urlbase: str | None = None, auth_token: str | N # Try to 'open' the remote path try: - self.meta = info(self.path, self.urlbase, auth_token=self.auth_token) + self.meta = info( + self.path, + self.urlbase, + auth_token=self.auth_token, + traffic=self.traffic, + ) except _httpx().HTTPStatusError as err: # HTTPStatusError only (not the broader HTTPError, which also covers # connection-level failures): a 404 means "not found", a connection @@ -1163,7 +1169,12 @@ def _reread_meta(self) -> None: """ with self._meta_lock: seen = self._meta_epoch - meta = info(self.path, self.urlbase, auth_token=self.auth_token) + meta = info( + self.path, + self.urlbase, + auth_token=self.auth_token, + traffic=self.traffic, + ) with self._meta_lock: if self._meta_epoch != seen: return diff --git a/src/blosc2/proxy.py b/src/blosc2/proxy.py index c2b1160e7..61fb73c93 100644 --- a/src/blosc2/proxy.py +++ b/src/blosc2/proxy.py @@ -251,10 +251,9 @@ def traffic(self) -> "blosc2.proxy_source.Traffic | None": """What this proxy has read off its source, or None for a local one. Cumulative bytes and requests since the source was opened, counted at the - transport, so the frame index and the block offsets are in it as well as - the data, and the metadata call that opened the handle is not. What a - slice cost in traffic is the difference between two readings of this, or - one reading after :meth:`Traffic.reset`. + transport, including metadata, frame indexes, block offsets, and data. + What a slice cost in traffic is the difference between two readings of + this, or one reading after :meth:`Traffic.reset`. It is what says whether block granularity is doing anything for a given dataset and access pattern: whole chunks and blocks of them take similar diff --git a/tests/ndarray/test_c2array_blocks.py b/tests/ndarray/test_c2array_blocks.py index cf11c85a5..8511f5f20 100644 --- a/tests/ndarray/test_c2array_blocks.py +++ b/tests/ndarray/test_c2array_blocks.py @@ -955,14 +955,16 @@ def test_traffic_counts_what_crossed_the_wire(server, any_chunk_wants_blocks): array, srv = server(data, chunks=(100, 200), blocks=(10, 20)) p = blosc2.Proxy(array, mode="w") assert p.traffic is array.traffic # the array's tally, not a second one + info_requests = sum(kind == "info" for kind, _, _ in srv.log) + assert p.traffic.requests == info_requests + assert p.traffic.nbytes == _bytes(srv, "info") p.traffic.reset() assert np.array_equal(p[0:5, 0:10], data[0:5, 0:10]) blocks = (p.traffic.requests, p.traffic.nbytes) assert blocks[0] > 0 assert blocks[1] > 0 - # What the server logged for the data endpoints is what was counted; the - # `api/info` that opened the handle is metadata and is deliberately not + # After the reset, what the server logged for the data endpoints is what was counted. served = [(kind, nbytes) for kind, _, nbytes in srv.log if kind != "info"] assert blocks[0] == len(served) assert blocks[1] <= sum(nbytes for _, nbytes in served) From 5200b647b5eee5109693e9c86e6a7964af90496f Mon Sep 17 00:00:00 2001 From: Francesc Alted Date: Fri, 4 Sep 2026 11:50:00 +0200 Subject: [PATCH 08/10] Report persistent cache status --- examples/fsspec-cat2-access.py | 5 ++--- src/blosc2/proxy.py | 11 +++++++++++ src/blosc2/schunk.py | 21 +++++++++++++-------- tests/test_fsspec.py | 3 +++ 4 files changed, 29 insertions(+), 11 deletions(-) diff --git a/examples/fsspec-cat2-access.py b/examples/fsspec-cat2-access.py index 4f685a0fd..547c963d6 100644 --- a/examples/fsspec-cat2-access.py +++ b/examples/fsspec-cat2-access.py @@ -48,8 +48,6 @@ def size_text(size: int) -> str: def benchmark(label: str, urlpath, cache_dir: Path) -> np.ndarray: - cache_existed = cache_dir.is_dir() and any(cache_dir.glob("*.b2nd")) - start = perf_counter() array = blosc2.open(urlpath, lazy=True, cache_dir=cache_dir) open_time = perf_counter() - start @@ -57,6 +55,7 @@ def benchmark(label: str, urlpath, cache_dir: Path) -> np.ndarray: metadata = (array.shape, array.dtype, array.chunks, array.blocks) cache_path = Path(array.urlpath).resolve() + cache_status = array.cache_status array.traffic.reset() start = perf_counter() @@ -83,7 +82,7 @@ def benchmark(label: str, urlpath, cache_dir: Path) -> np.ndarray: print(f"\n{label}") print(f" metadata: shape={metadata[0]}, dtype={metadata[1]}") print(f" chunks={metadata[2]}, blocks={metadata[3]}") - print(f" persistent cache: {cache_path} ({'existing' if cache_existed else 'new'})") + print(f" persistent cache: {cache_path} ({cache_status})") print(f" {'open + remote metadata:':<27}{open_time * 1000:.0f} ms ({open_traffic})") print(f" {'first data slice:':<27}{first_read_time * 1000:.0f} ms ({first_traffic})") print(f" {'cache after slice:':<27}{size_text(cache_size)}") diff --git a/src/blosc2/proxy.py b/src/blosc2/proxy.py index 61fb73c93..0c14a40b2 100644 --- a/src/blosc2/proxy.py +++ b/src/blosc2/proxy.py @@ -128,6 +128,7 @@ def __init__( """ self.src = src self.urlpath = urlpath + self._cache_status = None if kwargs is None: kwargs = {} self._cache = kwargs.pop("_cache", None) @@ -274,6 +275,16 @@ def traffic(self) -> "blosc2.proxy_source.Traffic | None": """ return getattr(self.src, "traffic", None) + @property + def cache_status(self) -> str | None: + """How the persistent cache was handled when this proxy was opened. + + This is ``"created"``, ``"reused"``, or ``"invalidated/rebuilt"`` for + a remote proxy opened with ``cache_dir`` or ``cache_path``. It is + ``None`` for proxies without a managed persistent cache. + """ + return self._cache_status + def __enter__(self) -> "Proxy": """Enter a context manager and return this proxy.""" return self diff --git a/src/blosc2/schunk.py b/src/blosc2/schunk.py index 7fe4925d7..928e1f413 100644 --- a/src/blosc2/schunk.py +++ b/src/blosc2/schunk.py @@ -1896,9 +1896,7 @@ def process_opened_object(res): src = blosc2.FsspecNDSource(proxy_src["urlpath"]) return blosc2.Proxy(src, _cache=proxy_cache, _refresh_source=False) if source_kind == "caterva2": - src = blosc2.C2Array( - proxy_src["urlpath"][0], proxy_src["urlpath"][1], proxy_src["urlpath"][2] - ) + src = blosc2.C2Array(proxy_src["urlpath"][0], proxy_src["urlpath"][1], proxy_src["urlpath"][2]) return blosc2.Proxy(src, _cache=proxy_cache, _refresh_source=False) if proxy_src["local_abspath"] is not None: source_path = proxy_src["local_abspath"] @@ -2049,13 +2047,20 @@ def _lazy_remote_proxy( else: path = fsspec_cache_path(identity, cache_dir, ".b2nd") stamp = getattr(src, "stamp", None) - if os.path.exists(path) and _cache_stamp(path) != stamp: - # The remote frame was replaced, which makes every cached chunk -- and - # every offset they were fetched by -- meaningless - blosc2.remove_urlpath(path) + cache_status = "created" + if os.path.exists(path): + if _cache_stamp(path) != stamp: + # The remote frame was replaced, which makes every cached chunk -- and + # every offset they were fetched by -- meaningless + blosc2.remove_urlpath(path) + cache_status = "invalidated/rebuilt" + else: + cache_status = "reused" # Proxy stamps the cache with src.stamp itself, and refuses one built against # other bytes; removing it above is what turns that refusal into a refetch - return blosc2.Proxy(src, urlpath=path, mode="a", _refresh_source=not source_fresh) + proxy = blosc2.Proxy(src, urlpath=path, mode="a", _refresh_source=not source_fresh) + proxy._cache_status = cache_status + return proxy def _open_c2_urlpath(urlpath: blosc2.URLPath, mode: str, offset: int, kwargs: dict): diff --git a/tests/test_fsspec.py b/tests/test_fsspec.py index 29d41d8d5..a1b5f63b9 100644 --- a/tests/test_fsspec.py +++ b/tests/test_fsspec.py @@ -493,12 +493,14 @@ def test_lazy_with_cache_dir(tmp_path, monkeypatch): ) p = blosc2.open(url, lazy=True, cache_dir=tmp_path) + assert p.cache_status == "created" assert np.array_equal(p[0:100], a[0:100]) assert fetched == [0] del p # A later run starts from the chunks the previous one pulled p = blosc2.open(url, lazy=True, cache_dir=tmp_path) + assert p.cache_status == "reused" assert np.array_equal(p[0:100], a[0:100]) assert fetched == [0] assert np.array_equal(p[500:600], a[500:600]) @@ -577,6 +579,7 @@ def test_lazy_cache_rebuilt_when_remote_changes(tmp_path): # they were fetched by, so the cache must be thrown away rather than reused _put("lazystale.b2nd", b) p = blosc2.open(url, lazy=True, cache_dir=tmp_path) + assert p.cache_status == "invalidated/rebuilt" assert np.array_equal(p[0:100], b[0:100]) From 408899dc1e959940168b53904ac439d8a0cf61d9 Mon Sep 17 00:00:00 2001 From: Francesc Alted Date: Fri, 4 Sep 2026 11:54:33 +0200 Subject: [PATCH 09/10] Update C2Array test doubles for traffic --- tests/ndarray/test_c2array_async.py | 1 + tests/test_b2objects.py | 2 +- tests/test_objectarray.py | 2 +- 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/ndarray/test_c2array_async.py b/tests/ndarray/test_c2array_async.py index 4556eaa5f..a11887680 100644 --- a/tests/ndarray/test_c2array_async.py +++ b/tests/ndarray/test_c2array_async.py @@ -18,6 +18,7 @@ class _FakeResponse: def __init__(self, json_data): self._json = json_data + self.content = b"" def raise_for_status(self): pass diff --git a/tests/test_b2objects.py b/tests/test_b2objects.py index 6317d34e6..08d6a52a2 100644 --- a/tests/test_b2objects.py +++ b/tests/test_b2objects.py @@ -32,7 +32,7 @@ def _make_c2array( ): dtype = np.dtype(dtype) - def fake_info(path_, urlbase_, params=None, headers=None, model=None, auth_token=None): + def fake_info(path_, urlbase_, params=None, headers=None, model=None, auth_token=None, traffic=None): return { "shape": list(shape), "chunks": list(chunks), diff --git a/tests/test_objectarray.py b/tests/test_objectarray.py index f86a2c31b..53c526c59 100644 --- a/tests/test_objectarray.py +++ b/tests/test_objectarray.py @@ -58,7 +58,7 @@ def _make_nested_blosc2_objects(): def _make_c2array(monkeypatch, path="@public/examples/ds-1d.b2nd", urlbase="https://cat2.cloud/demo/"): - def fake_info(path_, urlbase_, params=None, headers=None, model=None, auth_token=None): + def fake_info(path_, urlbase_, params=None, headers=None, model=None, auth_token=None, traffic=None): return {"schunk": {"cparams": dict(blosc2.cparams_dflts)}} monkeypatch.setattr(blosc2_c2array, "info", fake_info) From 47fd16514bdee06550ab41ef4874890fb64586de Mon Sep 17 00:00:00 2001 From: Francesc Alted Date: Fri, 4 Sep 2026 12:12:49 +0200 Subject: [PATCH 10/10] Reduce default test suite runtime --- tests/ctable/test_ctable_indexing.py | 6 ++-- tests/ctable/test_dictionary_column.py | 39 ++++++++------------------ tests/ctable/test_utf8.py | 1 + tests/test_ctable_cframe.py | 3 +- tests/test_locking.py | 10 +++---- 5 files changed, 23 insertions(+), 36 deletions(-) diff --git a/tests/ctable/test_ctable_indexing.py b/tests/ctable/test_ctable_indexing.py index c4dae75a1..dd3e00604 100644 --- a/tests/ctable/test_ctable_indexing.py +++ b/tests/ctable/test_ctable_indexing.py @@ -848,7 +848,7 @@ class _IncrRow: i: int = blosc2.field(blosc2.int64(), chunks=(2000,), blocks=(500,)) -def _build_incr_data(n=9000): +def _build_incr_data(n=2250): rng = np.random.default_rng(7) f = (rng.standard_normal(n) * 50).astype(np.float32) f[rng.integers(0, n, n // 100)] = np.nan # exercise NaN flags @@ -867,6 +867,7 @@ def _summary_sidecars(table): return out +@pytest.mark.heavy def test_incremental_summary_matches_ooc_build(tmp_path): """The incremental per-block accumulator (folded during the write phase) must produce SUMMARY sidecars byte-identical to the out-of-core @@ -901,10 +902,11 @@ def test_incremental_summary_matches_ooc_build(tmp_path): assert np.allclose(a["max"], b["max"], equal_nan=True) +@pytest.mark.heavy def test_incremental_summary_stale_on_inplace(tmp_path): """An in-place column write before close must invalidate the accumulator so the builder falls back to a correct full rescan.""" - f, i = _build_incr_data(n=4000) + f, i = _build_incr_data(n=2250) path = str(tmp_path / "upd.b2z") with blosc2.CTable(_IncrRow, urlpath=path, mode="w") as t: t.extend({"f": f, "i": i}) diff --git a/tests/ctable/test_dictionary_column.py b/tests/ctable/test_dictionary_column.py index f7cf95bb0..da53396a5 100644 --- a/tests/ctable/test_dictionary_column.py +++ b/tests/ctable/test_dictionary_column.py @@ -600,7 +600,6 @@ def test_dictionary_column_comparisons_are_elementwise(): It used to return a plain ``False`` — silently wrong rather than an error. """ - import numpy as np @dataclass class Row: @@ -625,7 +624,6 @@ def test_dictionary_ne_predicate_matches_live_rows(): Negating afterwards turned every dead capacity slot True, which then failed with an IndexError when used to select rows. """ - import numpy as np @dataclass class Row: @@ -637,11 +635,9 @@ class Row: t._flush_varlen_columns() assert sorted(t[t["c"] != "a1"]["c"][:]) == sorted(v for v in values if v != "a1") - assert len(t[t["c"] == "a1"]["c"][:]) == 13 - # A value no row carries: nothing matches, everything differs. - assert len(t[t["c"] == "absent"]["c"][:]) == 0 + # A value no row carries differs from every live row, but not from padded + # capacity slots. assert len(t[t["c"] != "absent"]["c"][:]) == len(values) - assert np.asarray((t["c"] != "a1")[:]).sum() == 26 def test_dictionary_index_answers_equality(tmp_path): @@ -652,27 +648,16 @@ class Row: c: str = blosc2.field(blosc2.dictionary()) values = ["pear", "apple", "cherry", "apple", "banana"] - results = {} - for tag in ("scan", "index"): - t = CTable(Row, urlpath=str(tmp_path / f"{tag}.b2t"), mode="w") - t.extend({"c": values}, validate=False) - t._flush_varlen_columns() - if tag == "index": - t.create_index("c", kind="full") - assert t["c"]._dictionary_index_mask("apple") is not None - # A value absent from the dictionary still answers, matching nothing. - assert not t["c"]._dictionary_index_mask("absent").any() - results[tag] = { - probe: ( - sorted(t[t["c"] == probe]["c"][:]), - sorted(t[t["c"] != probe]["c"][:]), - ) - for probe in ("apple", "pear", "absent") - } - del t - - assert results["index"] == results["scan"] - assert results["scan"]["apple"][0] == ["apple", "apple"] + t = CTable(Row, urlpath=str(tmp_path / "indexed.b2t"), mode="w") + t.extend({"c": values}, validate=False) + t._flush_varlen_columns() + t.create_index("c", kind="full") + assert t["c"]._dictionary_index_mask("apple") is not None + # A value absent from the dictionary still answers, matching nothing. + assert not t["c"]._dictionary_index_mask("absent").any() + assert sorted(t[t["c"] == "apple"]["c"][:]) == ["apple", "apple"] + assert list(t[t["c"] == "absent"]["c"][:]) == [] + del t def test_dictionary_index_spans_deleted_rows(tmp_path): diff --git a/tests/ctable/test_utf8.py b/tests/ctable/test_utf8.py index d4b6af359..33ef053c9 100644 --- a/tests/ctable/test_utf8.py +++ b/tests/ctable/test_utf8.py @@ -2299,6 +2299,7 @@ def test_constructors_string_dtype_reject_nd(): blosc2.zeros(3, dtype=STRING_DTYPE, urlpath="unused.b2nd") +@pytest.mark.heavy @pytest.mark.skipif( blosc2.IS_WASM, reason="peak-memory scaling is not measurable under Pyodide: its noise floor " diff --git a/tests/test_ctable_cframe.py b/tests/test_ctable_cframe.py index cc1205e95..c072a0fa5 100644 --- a/tests/test_ctable_cframe.py +++ b/tests/test_ctable_cframe.py @@ -203,8 +203,7 @@ class R: p = pathlib.Path(tempfile.mkdtemp()) / "t.b2z" t = blosc2.CTable(R, urlpath=str(p), mode="w", compact=True) - for i in range(50): - t.append((i, f"n{i}")) + t.extend([(i, f"n{i}") for i in range(50)]) t.close() t = blosc2.open(p) cf = t.to_cframe() diff --git a/tests/test_locking.py b/tests/test_locking.py index 16dbf05a0..86718bd37 100644 --- a/tests/test_locking.py +++ b/tests/test_locking.py @@ -225,7 +225,7 @@ def test_cross_process_hammer(tmp_path): urlpath = tmp_path / "schunk-hammer.b2frame" create_schunk(urlpath, contiguous=False, locking=True) - iters = 500 + iters = 150 writer = subprocess.Popen( [sys.executable, "-c", WRITER_SCRIPT, str(urlpath), str(NCHUNKS), str(CHUNK_NITEMS), str(iters)] ) @@ -353,7 +353,7 @@ def test_cross_process_multiwriter_update(tmp_path): # owner's last-written value. urlpath = tmp_path / "schunk-multiwriter-update.b2frame" nwriters = 4 - iters = 60 + iters = 20 schunk = create_schunk(urlpath, contiguous=False, locking=True) nchunks = schunk.nchunks del schunk @@ -1332,10 +1332,10 @@ def slow_sync(self): dstore = blosc2.DictStore(path, mode="w", threshold=500, locking=True) dstore["/hot"] = np.arange(100) - writer = subprocess.Popen([sys.executable, "-c", DSTORE_OVERWRITER, path, "300"]) + writer = subprocess.Popen([sys.executable, "-c", DSTORE_OVERWRITER, path, "100"]) try: nreads = 0 - while writer.poll() is None and nreads < 60: + while writer.poll() is None and nreads < 20: data = dstore["/hot"][:] # Each round writes arange(i, i + 100); a torn read breaks the run assert np.array_equal(data, np.arange(data[0], data[0] + 100)) @@ -1345,7 +1345,7 @@ def slow_sync(self): writer.kill() writer.wait() - assert nreads == 60 + assert nreads == 20 dstore._closed = True