diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index cca6a13af..f7ead21ff 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -6,6 +6,36 @@ XXX version-specific blurb XXX ### Improvements +* New `blosc2[fsspec]` extra: `blosc2.open()`, `save_array()` and `save_tensor()` + accept any [fsspec](https://filesystem-spec.readthedocs.io) URL — `s3://`, + `gs://`, `zip://`, chained ones like `zip://inner.b2nd::s3://bucket/a.zip`. + `open()` reads the container whole, or through a staleness-checked local copy + with `cache_storage=` (which is what covers `.b2d` stores, sparse frames, + `offset` and `mmap_mode`), or one chunk at a time with `lazy=True`, which + leaves a huge frame where it is and fetches only the chunks a slice touches + through the new `blosc2.FsspecNDSource`. The two combine: `lazy=True` with a + `cache_storage=` keeps the fetched chunks there, so a later run starts from + them. Protocol drivers (`s3fs`, `gcsfs`...) and credentials stay the caller's + business. On the write side `NDArray.save()` and `blosc2.save()` upload the + whole array as one object; containers cannot be *backed* by a URL while they + are written (the C layer rewrites a frame's header and offsets as chunks land, + which an object store has no way to serve), so constructors given a URL now say + that instead of failing deep in C. + +* `Proxy.fetch()` takes a `max_concurrency=` argument, and reads it from the + source when the source has one, so `blosc2.open(url, lazy=True, + max_concurrency=...)` overlaps its chunk fetches in a thread pool. Ordinary + slicing benefits, not just the async `afetch()`. A lazy fsspec proxy defaults + to 8, matching what `afetch()` already used for remote sources; pass 1 for a + protocol with no latency to hide, where the pool costs ~10 µs per chunk and + saves nothing. Other sources stay serial unless asked, since this is only safe + for a thread-safe `get_chunk`. + +* `blosc2.Proxy(src, urlpath=..., mode="a")` now adopts the cache left by an + earlier run instead of failing on the existing file, so a proxy's cache can + outlive the process. The cache must come from a proxy over a source of the same + shape and dtype; anything else at that path raises. + * Querying a `utf8()` column through its FULL index no longer materializes the index vocabulary. The query literal is turned into an alphabetical rank by bisecting the vocabulary sidecar instead, so a lookup reads a few blocks diff --git a/bench/ndarray/fsspec-block-granularity.py b/bench/ndarray/fsspec-block-granularity.py new file mode 100644 index 000000000..66d14e635 --- /dev/null +++ b/bench/ndarray/fsspec-block-granularity.py @@ -0,0 +1,278 @@ +#!/usr/bin/env python + +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +"""Would block-granular downloads beat chunk-granular ones for a given array? + +``blosc2.open(url, lazy=True)`` fetches one whole compressed chunk per range +request. A chunk is made of blocks, which blosc2 compresses and decompresses +independently, so a slice could in principle fetch only the blocks it touches. +Whether that is worth the extra round trip it costs (the block offsets live in +the chunk header, which has to be read first) depends on two numbers this script +measures: + +- the **touch ratio**: what fraction of the bytes of the chunks a slice touches + its blocks actually account for. Exact, computed locally from the array's own + chunk headers, no network involved; +- the **wall time** of the two request patterns against a real object store. + +Usage +----- + # touch ratios only, on any local array + python fsspec-block-granularity.py mydata.b2nd + + # ... and time both request patterns against real S3 + python fsspec-block-granularity.py mydata.b2nd \\ + --replay s3://noaa-goes16/ABI-L1b-RadF/2020/001/00/OR_ABI-L1b-RadF-M6C02_G16_s20200010000216_e20200010009524_c20200010009570.nc --anon + +The replay target is *any* object at least as large as the biggest request; its +contents are never used. What is being timed is the request shape — how many +ranges, of what sizes, in how many dependent phases — which is what separates +the two designs. Using a public object means the measurement needs no bucket of +its own, and the client stack (s3fs, aiobotocore, HTTPS, real latency) is the +one blosc2 would use. + +Three modes are timed: + +- ``chunk``: one range per touched chunk, ``max_concurrency`` at a time. What + ``lazy=True`` does today. +- ``blocks``: one range per touched chunk for the header and block offsets, + then one range per (coalesced) run of wanted blocks. Two dependent phases. +- ``blocks, cached``: the same without the header phase, which is what a second + slice of the same array costs once the offsets have been read once. +""" + +import argparse +import itertools +import math +import random +import statistics +import time +from concurrent.futures import ThreadPoolExecutor + +import numpy as np + +import blosc2 + +GAP = 4096 # merge ranges separated by less than this into one request + + +def chunk_layout(schunk, nchunk, cache): + """(cbytes, bstarts, extents) of a chunk, as a byte-range reader would see it. + + ``bstarts`` is *not* sorted -- a multithreaded compressor writes blocks in + completion order -- so a block's extent is the distance to the next larger + offset, not to its neighbour in the array. The extents are computed that way + here, rather than read from the lazy chunk's trailer, because that is all a + byte-range reader over the network can do: it is an upper bound where a chunk + has holes, which is what such a reader would fetch. + """ + if nchunk in cache: + return cache[nchunk] + # A lazy chunk is header + bstarts + trailer, so this reads a few hundred + # bytes per chunk instead of the whole array + chunk = schunk.get_lazychunk(nchunk) + nbytes, cbytes, blocksize = blosc2.get_cbuffer_sizes(chunk) + nblocks = (nbytes + blocksize - 1) // blocksize + if (chunk[31] >> 4) & 0x7: # run-length chunk: no bytes in the file at all + res = (0, np.empty(0, np.int64), np.empty(0, np.int64)) + else: + if chunk[2] & 0x02: # memcpyed: raw blocks, no bstarts section + bstarts = 32 + np.arange(nblocks, dtype=np.int64) * blocksize + extents = np.full(nblocks, blocksize, dtype=np.int64) + extents[-1] = nbytes - (nblocks - 1) * blocksize + else: + if len(chunk) < 32 + 4 * nblocks: # an in-memory array: no lazy chunks + chunk = schunk.get_chunk(nchunk) + bstarts = np.frombuffer(chunk[32 : 32 + 4 * nblocks], dtype="= 0 else index + shape[dim] + stop = start + 1 + spans.append((start, stop)) + chunk_grid = [math.ceil(s / c) for s, c in zip(shape, chunks, strict=True)] + blocks_in_chunk = [math.ceil(c / b) for c, b in zip(chunks, blocks, strict=True)] + out = {} + ranges = [range(s // chunks[d], (e - 1) // chunks[d] + 1) for d, (s, e) in enumerate(spans)] + for coords in itertools.product(*ranges): + nchunk = int(np.ravel_multi_index(coords, chunk_grid)) + per_dim = [] + for dim in range(ndim): + lo = max(spans[dim][0] - coords[dim] * chunks[dim], 0) + hi = min(spans[dim][1] - coords[dim] * chunks[dim], chunks[dim]) + per_dim.append(range(lo // blocks[dim], (hi - 1) // blocks[dim] + 1)) + out[nchunk] = [int(np.ravel_multi_index(b, blocks_in_chunk)) for b in itertools.product(*per_dim)] + return out + + +def request_plan(array, item): + """The requests each mode would issue for *item*: (chunk sizes, header sizes, block sizes).""" + schunk, cache = array.schunk, {} + chunk_sizes, header_sizes, block_sizes = [], [], [] + nblocks_touched = 0 + for nchunk, nblocks in touched(array.shape, array.chunks, array.blocks, item).items(): + cbytes, bstarts, extents = chunk_layout(schunk, nchunk, cache) + if not cbytes: # special chunk: free in both modes + continue + chunk_sizes.append(int(cbytes)) + header_sizes.append(32 + 4 * len(bstarts)) + nblocks_touched += len(nblocks) + block_sizes += coalesce([(int(bstarts[i]), int(extents[i])) for i in nblocks]) + return chunk_sizes, header_sizes, block_sizes, nblocks_touched + + +def default_patterns(shape): + """Slices worth asking about, for an array of any shape.""" + mid = [s // 2 for s in shape] + point = tuple(mid) + line_last = (*mid[:-1], slice(None)) + line_first = (slice(None), *mid[1:]) + window = tuple(slice(m, m + max(1, s // 64)) for m, s in zip(mid, shape, strict=True)) + slab = (slice(mid[0], mid[0] + max(1, shape[0] // 100)), *[slice(None)] * (len(shape) - 1)) + big_slab = (slice(mid[0], mid[0] + max(1, shape[0] // 10)), *[slice(None)] * (len(shape) - 1)) + return [ + ("point", point), + ("line, last dim", line_last), + ("line, first dim", line_first), + ("window (1/64 per dim)", window), + ("slab (1% of dim 0)", slab), + ("slab (10% of dim 0)", big_slab), + ] + + +class Replayer: + """Issues the request pattern of a plan against a real object store.""" + + def __init__(self, urlpath, concurrency, anon=False, endpoint_url=None): + import fsspec + + options = {k: v for k, v in {"anon": anon, "endpoint_url": endpoint_url}.items() if v} + if options: + fsspec.config.conf.setdefault(urlpath.split("://", 1)[0], {}).update(options) + self.fs, self.path = fsspec.url_to_fs(urlpath) + self.size = self.fs.info(self.path)["size"] + self.concurrency = concurrency + self.random = random.Random(7) + + def _one(self, size): + # A fresh offset every time, so nothing is served from a cache anywhere + size = min(size, self.size) + offset = self.random.randrange(0, self.size - size + 1) + return len(self.fs.cat_file(self.path, start=offset, end=offset + size)) + + def phase(self, sizes): + """One wave of parallel range reads, as Proxy.fetch issues them.""" + if not sizes: + return + with ThreadPoolExecutor(max_workers=min(self.concurrency, len(sizes))) as pool: + list(pool.map(self._one, sizes)) + + def time(self, phases): + t0 = time.perf_counter() + for sizes in phases: + self.phase(sizes) + return time.perf_counter() - t0 + + +def main(): + p = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + p.add_argument("urlpath", help="a local .b2nd array to take the geometry from") + p.add_argument("--replay", help="URL of any large object to replay the request pattern against") + p.add_argument("--anon", action="store_true", help="anonymous access to the replay target") + p.add_argument("--endpoint-url", help="for S3-compatible endpoints (R2, B2, MinIO...)") + p.add_argument("--concurrency", type=int, default=8, help="parallel requests (default: 8)") + p.add_argument("--reps", type=int, default=5, help="timed repetitions (default: 5)") + p.add_argument("--max-mb", type=float, default=45, help="skip patterns fetching more than this") + args = p.parse_args() + + array = blosc2.open(args.urlpath) + blocks_per_chunk = math.prod([math.ceil(c / b) for c, b in zip(array.chunks, array.blocks, strict=True)]) + print( + f"{args.urlpath}: shape={array.shape} dtype={array.dtype} chunks={array.chunks} " + f"blocks={array.blocks}\n {array.schunk.nchunks} chunks, {blocks_per_chunk} blocks/chunk, " + f"cratio {array.schunk.cratio:.1f}x" + ) + + plans = [] + print( + f"\n {'pattern':22s} {'chunks':>6s} {'blocks':>13s} {'chunk mode':>18s} {'block mode':>18s} ratio" + ) + for name, item in default_patterns(array.shape): + chunk_sizes, header_sizes, block_sizes, nblocks = request_plan(array, item) + chunk_bytes, block_bytes = sum(chunk_sizes), sum(header_sizes) + sum(block_sizes) + ratio = block_bytes / chunk_bytes if chunk_bytes else float("nan") + print( + f" {name:22s} {len(chunk_sizes):6d} {nblocks:6d}/{len(chunk_sizes) * blocks_per_chunk:<6d} " + f"{len(chunk_sizes):5d} req {chunk_bytes / 1e6:7.2f} MB " + f"{len(header_sizes) + len(block_sizes):5d} req {block_bytes / 1e6:7.2f} MB {ratio * 100:6.1f}%" + ) + plans.append((name, chunk_sizes, header_sizes, block_sizes, chunk_bytes, block_bytes)) + + if not args.replay: + return + + replayer = Replayer(args.replay, args.concurrency, args.anon, args.endpoint_url) + print( + f"\nreplaying against {args.replay} ({replayer.size / 1e6:.0f} MB), " + f"concurrency {args.concurrency}, {args.reps} reps" + ) + times = {name: {"chunk": [], "blocks": [], "cached": []} for name, *_ in plans} + for rep in range(args.reps): + for name, chunk_sizes, header_sizes, block_sizes, chunk_bytes, _ in plans: + if chunk_bytes > args.max_mb * 1e6: + continue + times[name]["chunk"].append(replayer.time([chunk_sizes])) + times[name]["blocks"].append(replayer.time([header_sizes, block_sizes])) + times[name]["cached"].append(replayer.time([block_sizes])) + print(f" rep {rep + 1}/{args.reps}", flush=True) + + print(f"\n {'pattern':22s} {'chunk mode':>16s} {'blocks':>17s} {'blocks, cached':>17s}") + for name, _sizes, _, _, chunk_bytes, block_bytes in plans: + if not times[name]["chunk"]: + print(f" {name:22s} skipped ({chunk_bytes / 1e6:.0f} MB > --max-mb)") + continue + median = {k: statistics.median(v) for k, v in times[name].items()} + print( + f" {name:22s} {chunk_bytes / 1e6:6.2f}MB {median['chunk']:5.2f}s " + f"{block_bytes / 1e6:6.2f}MB {median['blocks']:5.2f}s {median['chunk'] / median['blocks']:4.1f}x " + f"{median['cached']:11.2f}s {median['chunk'] / median['cached']:4.1f}x" + ) + + +if __name__ == "__main__": + main() diff --git a/bench/ndarray/fsspec-concurrency.py b/bench/ndarray/fsspec-concurrency.py new file mode 100644 index 000000000..65a0ba00d --- /dev/null +++ b/bench/ndarray/fsspec-concurrency.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python + +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +"""Find the right ``max_concurrency`` for lazy reads against a real object store. + +``blosc2.open(url, lazy=True)`` fetches one chunk per range request and overlaps +8 of them by default. That 8 was chosen by argument, not measurement: the cost +of the thread pool was measured (~10 us per chunk, where there is no latency to +hide) but the gain never was, because nothing that runs offline has a round trip +to hide. This script answers it against a real endpoint. + +It also runs ``afetch()``, which is worth keeping in the sweep: the first time +this script was pointed at a real S3 endpoint, that path failed outright, and +``memory://`` cannot reproduce it because it is not an async backend. + +Usage +----- + python fsspec-concurrency.py s3://my-bucket/bench.b2nd + + # S3-compatible endpoints (Cloudflare R2, Backblaze B2, MinIO...) + python fsspec-concurrency.py s3://my-bucket/bench.b2nd \\ + --endpoint-url https://.r2.cloudflarestorage.com + +The target array is written on first use, so the bucket must be writable; pass +``--no-upload`` to benchmark one that is already there. Credentials come from +the driver (``~/.aws/config``, ``AWS_*`` environment variables, ...), never from +blosc2. + +``--endpoint-url`` and ``--anon`` go through ``fsspec.config.conf``, which is +fsspec's own per-protocol default-argument mechanism, because ``blosc2.open()`` +has no ``storage_options=`` passthrough of its own. The equivalent without this +script is a ``~/.config/fsspec/s3.json`` holding ``{"anon": true}``, or the +``FSSPEC_S3_ANON`` / ``FSSPEC_S3_ENDPOINT_URL`` environment variables. + +Checking it works, without an account +------------------------------------- +``moto[server]`` gives a local S3 endpoint over real HTTP, which is enough to +prove the async path runs at all before spending anything on a real bucket:: + + pip install "moto[server]" + moto_server -p 5000 & + export AWS_ACCESS_KEY_ID=x AWS_SECRET_ACCESS_KEY=x AWS_DEFAULT_REGION=us-east-1 + python -c "import s3fs; s3fs.S3FileSystem(endpoint_url='http://127.0.0.1:5000').mkdir('bench')" + python fsspec-concurrency.py s3://bench/bench.b2nd --endpoint-url http://127.0.0.1:5000 + +Do not read the timings from that run. moto is a single-process Python mock: +it has no network latency for concurrency to hide, and it may well serialize the +requests it receives, so the sweep can show a flat or inverted curve while the +client side is behaving perfectly. It answers "does this work", not "how fast". +MinIO is the better local endpoint if you want a genuinely concurrent server, +and neither substitutes for a real bucket on a real network. + +Reading the output +------------------ +Wall time should fall roughly as 1/concurrency while requests are the +bottleneck, then flatten once the endpoint, the connection pool, or the local +CPU becomes one. The knee is the answer. Times going *up* again, or errors +appearing, means the endpoint is throttling (S3 answers 503 SlowDown): back off +to the last value that scaled. +""" + +import argparse +import asyncio +import time + +import numpy as np + +import blosc2 + + +def configure(protocol, **options): + """Set fsspec's default arguments for *protocol*, since we cannot pass them.""" + options = {k: v for k, v in options.items() if v} + if options: + import fsspec.config + + fsspec.config.conf.setdefault(protocol, {}).update(options) + print(f"fsspec {protocol} options: {options}") + + +def build_target(urlpath, nchunks, chunklen, upload): + """Put an array of a known shape at *urlpath*, unless it is there already.""" + import fsspec + + fs, path = fsspec.url_to_fs(urlpath) + if fs.exists(path): + if not upload: + return + print(f"overwriting {urlpath}") + elif not upload: + raise SystemExit(f"{urlpath} does not exist and --no-upload was passed") + + a = blosc2.arange(0, nchunks * chunklen, dtype=np.int32, chunks=(chunklen,)) + t0 = time.perf_counter() + a.save(urlpath) + print(f"uploaded {a.schunk.nchunks} chunks in {time.perf_counter() - t0:.1f} s") + + +def timed(urlpath, item, max_concurrency, use_afetch=False): + """Time one cold read: a fresh proxy every time, so nothing is cached.""" + a = blosc2.open(urlpath, lazy=True, max_concurrency=max_concurrency) + t0 = time.perf_counter() + if use_afetch: + asyncio.run(a.afetch(item, max_concurrency=max_concurrency)) + else: + a[item] + return time.perf_counter() - t0 + + +def report(label, times, nchunks_touched): + print(f"\n{label} ({nchunks_touched} chunks)") + print(f" {'concurrency':>11} {'wall':>8} {'per chunk':>10} {'vs serial':>9}") + serial = times.get(1) + for concurrency, elapsed in sorted(times.items()): + speedup = f"{serial / elapsed:.1f}x" if serial else "-" + print( + f" {concurrency:>11} {elapsed:>7.2f}s {elapsed / nchunks_touched * 1e3:>9.1f}ms {speedup:>9}" + ) + + +def main(): + p = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + p.add_argument("urlpath", help="fsspec URL of the benchmark array, e.g. s3://bucket/bench.b2nd") + p.add_argument("--endpoint-url", help="for S3-compatible endpoints (R2, B2, MinIO...)") + p.add_argument("--anon", action="store_true", help="anonymous access (public read-only buckets)") + p.add_argument("--nchunks", type=int, default=200, help="chunks in the uploaded array") + p.add_argument("--chunklen", type=int, default=250_000, help="int32 items per chunk (1 MB each)") + p.add_argument("--no-upload", dest="upload", action="store_false", help="use the array as it is") + p.add_argument( + "--concurrency", + default="1,2,4,8,16,32", + help="comma-separated values to try (default: 1,2,4,8,16,32)", + ) + p.add_argument("--skip-afetch", action="store_true", help="do not exercise the async path") + args = p.parse_args() + + protocol = args.urlpath.split("://", 1)[0] + configure(protocol, endpoint_url=args.endpoint_url, anon=args.anon) + build_target(args.urlpath, args.nchunks, args.chunklen, args.upload) + levels = [int(x) for x in args.concurrency.split(",")] + + # One chunk, serially: the round trip everything else is made of + one = timed(args.urlpath, slice(0, args.chunklen), 1) + print(f"\nsingle chunk (one round trip): {one * 1e3:.0f} ms") + + slice_chunks = min(16, args.nchunks) + item = slice(0, slice_chunks * args.chunklen) + report("slice", {c: timed(args.urlpath, item, c) for c in levels}, slice_chunks) + report( + "whole array", + {c: timed(args.urlpath, slice(None), c) for c in levels}, + args.nchunks, + ) + + if not args.skip_afetch: + # The async path, which only a real async backend exercises + report( + "slice, afetch (async path)", + {c: timed(args.urlpath, item, c, use_afetch=True) for c in levels}, + slice_chunks, + ) + + +if __name__ == "__main__": + main() diff --git a/doc/getting_started/installation.rst b/doc/getting_started/installation.rst index 1b7493837..4e96405b2 100644 --- a/doc/getting_started/installation.rst +++ b/doc/getting_started/installation.rst @@ -40,6 +40,12 @@ grouped into *extras* that you opt into with the ``blosc2[extra]`` syntax: * - ``parquet`` - The ``parquet-to-blosc2`` converter (``pyarrow``); see :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. Install one or more extras by listing them in brackets (quote the argument in shells like ``zsh`` that treat brackets specially): @@ -49,8 +55,17 @@ 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]" s3fs # fsspec URLs, plus the S3 driver 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 +one chunk at a time (``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, +and ``examples/ndarray/concurrent-fsspec.py`` shows what overlapping the chunk +fetches buys; both run with no network or credentials. + Source code +++++++++++ diff --git a/doc/reference/classes.rst b/doc/reference/classes.rst index 46c935617..1a33777fd 100644 --- a/doc/reference/classes.rst +++ b/doc/reference/classes.rst @@ -139,6 +139,7 @@ container APIs above. proxy proxysource proxyndsource + fsspecndsource simpleproxy embed_store dict_store diff --git a/doc/reference/fsspecndsource.rst b/doc/reference/fsspecndsource.rst new file mode 100644 index 000000000..399e0eb83 --- /dev/null +++ b/doc/reference/fsspecndsource.rst @@ -0,0 +1,22 @@ +.. _FsspecNDSource: + +FsspecNDSource +============== + +A :ref:`ProxyNDSource` 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. For other sources, see :ref:`ProxyNDSource` and +:ref:`ProxySource`. + +``examples/ndarray/rw-fsspec.py`` is a runnable walkthrough of this and the +other two ways to read an fsspec URL, and of writing one back. +``examples/ndarray/concurrent-fsspec.py`` measures ``max_concurrency`` against a +filesystem with a simulated round trip, since no protocol that runs offline has +latency for the thread pool to hide. + +.. currentmodule:: blosc2 + +.. autoclass:: FsspecNDSource + :members: + :exclude-members: all, any, max, mean, min, prod, std, sum, var + :member-order: groupwise diff --git a/examples/ndarray/concurrent-fsspec.py b/examples/ndarray/concurrent-fsspec.py new file mode 100644 index 000000000..ac5035545 --- /dev/null +++ b/examples/ndarray/concurrent-fsspec.py @@ -0,0 +1,92 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +# Concurrent chunk fetching for lazily-read remote NDArrays. +# +# Needs the fsspec extra: pip install "blosc2[fsspec]" +# +# blosc2.open(url, lazy=True) reads one chunk per range request, so a slice +# against an object store is almost entirely round-trip latency. Overlapping +# those requests (max_concurrency=) is what makes a wide slice bearable. +# +# Seeing that requires a filesystem that actually waits, and no protocol +# available offline has any latency to speak of: memory://, zip:// and tar:// +# are all local reads, where the thread pool can only lose (about 10 us per +# chunk). So this example bolts a fixed delay onto fsspec's in-memory +# filesystem to stand in for the network. Against a real s3:// bucket the +# delay is real and nothing else changes: +# +# a = blosc2.open("s3://my-bucket/big.b2nd", lazy=True) # 8 by default + +import time + +import fsspec +import numpy as np +from fsspec.implementations.memory import MemoryFileSystem + +import blosc2 + +ROUND_TRIP = 0.005 # 5 ms, a fast object store + + +class SlowMemoryFileSystem(MemoryFileSystem): + """fsspec's in-memory filesystem, with a network's worth of waiting.""" + + protocol = "slowmem" + + @classmethod + def _strip_protocol(cls, path): + if path.startswith("slowmem://"): + path = path[len("slowmem://") :] + return super()._strip_protocol(path) + + def cat_file(self, path, start=None, end=None, **kwargs): + time.sleep(ROUND_TRIP) + return super().cat_file(path, start, end, **kwargs) + + +fsspec.register_implementation("slowmem", SlowMemoryFileSystem) + +# 100 chunks. The store is shared with memory://, so we can write it fast and +# read it back slowly, which is what a remote array looks like anyway. +a = blosc2.arange(0, 1_000_000, dtype=np.int32, chunks=(10_000,)) +a.save("memory://big.b2nd") +print(f"array: {a.shape} in {a.schunk.nchunks} chunks, {ROUND_TRIP * 1e3:.0f} ms per fetch\n") + + +def timed(label, urlpath, item, **kwargs): + p = blosc2.open(urlpath, lazy=True, **kwargs) + t0 = time.perf_counter() + p[item] + elapsed = time.perf_counter() - t0 + print(f"{label:34s} {elapsed:5.2f} s") + return elapsed + + +# Reading the whole array: 100 fetches, serially or eight at a time +serial = timed("whole array, max_concurrency=1", "slowmem://big.b2nd", slice(None), max_concurrency=1) +default = timed("whole array, default (8)", "slowmem://big.b2nd", slice(None)) +print(f"{'':34s} {serial / default:5.1f}x faster\n") + +# A slice fetches only the chunks it touches, and those overlap too +serial = timed( + "12-chunk slice, max_concurrency=1", "slowmem://big.b2nd", slice(0, 120_000), max_concurrency=1 +) +default = timed("12-chunk slice, default (8)", "slowmem://big.b2nd", slice(0, 120_000)) +print(f"{'':34s} {serial / default:5.1f}x faster\n") + +# The cache means a chunk is only ever fetched once, so a repeat is free +p = blosc2.open("slowmem://big.b2nd", lazy=True) +p[0:120_000] +t0 = time.perf_counter() +p[0:120_000] +print(f"{'same slice again (cached)':34s} {time.perf_counter() - t0:5.2f} s") + +# On a protocol with no latency to hide, ask for serial: the pool costs about +# 10 us per chunk there and saves nothing +b = blosc2.open("memory://big.b2nd", lazy=True, max_concurrency=1) +np.testing.assert_array_equal(b[0:120_000], a[0:120_000]) diff --git a/examples/ndarray/rw-fsspec.py b/examples/ndarray/rw-fsspec.py new file mode 100644 index 000000000..3a8a0c524 --- /dev/null +++ b/examples/ndarray/rw-fsspec.py @@ -0,0 +1,66 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +# Writing and reading NDArrays through fsspec URLs. +# +# Needs the fsspec extra: pip install "blosc2[fsspec]" +# +# This uses memory:// so it runs anywhere, with no network and no credentials. +# Every URL below can be an s3://, gs://, abfs://... one instead, once the +# driver for that protocol is installed (s3fs, gcsfs, adlfs...): +# +# urlpath = "s3://my-bucket/ds-2d.b2nd" + +import tempfile + +import numpy as np + +import blosc2 + +urlpath = "memory://ds-2d.b2nd" + +a = blosc2.arange(0, 10_000, dtype=np.int32, shape=(100, 100), chunks=(10, 100)) + +# Write. The whole array goes up as a single object, replacing whatever was +# there: object stores have no partial write, so this is the only shape a +# remote write takes. Two writers to the same key silently lose one. +a.save(urlpath) + +# Read it back whole. This is one GET plus a rebuild in memory, which is the +# right thing for an array you are going to use all of. +b = blosc2.open(urlpath) +print(f"read whole: {type(b).__name__} {b.shape} {b.dtype}") +np.testing.assert_array_equal(b[:], a[:]) + +with tempfile.TemporaryDirectory() as cachedir: + # Read through a local cache. The container is downloaded once into + # cachedir and opened as an ordinary local path, so mmap, offsets and the + # directory formats (.b2d stores, sparse frames) all work, and a later run + # 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") + print(f"read cached: {c.shape} (mmapped from {cachedir})") + np.testing.assert_array_equal(c[:], a[:]) + + # Read lazily. Nothing is transferred up front: the array stays where it + # is and each slice fetches only the chunks it touches, one range request + # each. This is what you want for an array too big to download. + d = blosc2.open(urlpath, lazy=True, cache_storage=cachedir) + print(f"read lazy: {type(d).__name__} {d.shape} {d.dtype}") + + # Only the two chunks covering rows 15..25 are fetched here + np.testing.assert_array_equal(d[15:25], a[15:25]) + + # ...and they are cached, in cachedir, for the next run as well as this one + np.testing.assert_array_equal(d[15:25], a[15:25]) + + # A lazy handle is an ordinary operand, so expressions work on it, and + # slicing one still fetches only the chunks that slice needs + expr = d * 2 + print(f"lazy expression: {type(expr).__name__} -> {expr[15:17, 0]}") + np.testing.assert_array_equal(expr[15:25], a[15:25] * 2) diff --git a/plans/fsspec-blocks.md b/plans/fsspec-blocks.md new file mode 100644 index 000000000..6c2894891 --- /dev/null +++ b/plans/fsspec-blocks.md @@ -0,0 +1,451 @@ +# Block-Granular Downloads For `blosc2.open(url, lazy=True)` + +Analysis only — nothing implemented. Written 2026-08-16 on branch +`fsspec-support-plan`, after +[plans/fsspec-support.md](fsspec-support.md) phase 3 shipped. + +## The question + +`blosc2.open(url, lazy=True)` returns a `Proxy` over `FsspecNDSource` +([src/blosc2/proxy.py](../src/blosc2/proxy.py):817). A slice costs one range +request per chunk it touches, and each request pulls the **whole compressed +chunk** (`get_chunk`, :902, bounded by `_chunk_extents`, :802). With the +defaults `blosc2.asarray()` picks, a chunk is one to two orders of magnitude +larger than the smallest independently decompressable unit — the block. What +would it take to fetch blocks instead? + +**Verdict up front:** it is doable in pure Python, with no C changes, in about +300 lines, and the format cooperates better than expected — a chunk with only +some of its blocks present is a *valid* chunk that `update_chunk` accepts. It is +not free: it costs one extra round trip per fetch, which makes it a loss on +small chunks and on slices that want most of their blocks anyway. Measured +against real S3 (see below), that is **5–17x faster** on arrays with multi-MB +chunks, 2–5x on 1 MB chunks, and 0.5–0.7x on the rest — so the whole-chunk +fallback for chunks where blocks do not pay is part of the design, not a +refinement of it. + +## What chunk granularity costs today + +Measured on this machine, `blosc2.asarray()` defaults, `(2000, 2000)` f8: + +| | chunk | block | blocks/chunk | +|---|---|---|---| +| shape | (1000, 2000) | (8, 2000) | 125 | +| uncompressed | 16 MB | 128 KB | | +| compressed, `arange` data | 60 KB | ~1.5 KB | | +| compressed, random data | 6.7 MB | ~67 KB | | + +So a one-row slice downloads 16 MB worth of chunk to use 128 KB worth of block. +There is a second cost nobody has complained about yet: `Proxy.fetch` runs up to +`max_concurrency=8` chunk fetches at once, so peak memory is 8 × chunk cbytes — +54 MB for the random-data case above. Block granularity drops that to 8 × block +cbytes. + +## Format facts, verified rather than assumed + +All of these were checked against real chunks with +`blosc2.get_cbuffer_sizes()` + `blosc2.decompress2()` (script at the end). + +1. **Chunk header** is 32 bytes (`BLOSC_EXTENDED_HEADER_LENGTH`): `typesize` + @0x3, `nbytes` @0x4, `blocksize` @0x8, `cbytes` @0xC (int32 LE), `flags` @0x2 + (bit 1 = memcpyed, bit 4 = *no* split), `blosc2_flags` @0x1F (special kind in + bits 4-6, dict in bit 0), `flags2` @0x1E (bit 0 = variable-length blocks). + `nblocks = ceil(nbytes / blocksize)`. + +2. **`bstarts` is an `int32` array of `nblocks` entries right after the header**, + offsets relative to the chunk start. + +3. **`bstarts` is NOT monotonic.** This is the one that matters and it is the + opposite of what `README_CHUNK_FORMAT.rst` suggests to a casual reader (its + "compressed size is derived from adjacent entries" sentence is about + variable-length-block chunks only). A multithreaded compressor writes blocks + in completion order: + + ``` + nthreads=1 bstarts monotonic -> True + nthreads=4 bstarts monotonic -> False + nthreads=8 bstarts monotonic -> False, rank correlation with block index 0.99 + ``` + + Consequences: a block's extent is the distance to the **next larger** bstart + (sorted-neighbour, exactly the trick `_chunk_extents` already uses for + chunks), never `bstarts[i+1] - bstarts[i]`; and a contiguous run of blocks is + *nearly* but not exactly contiguous on the wire, so range coalescing has to + sort by offset and merge with a gap tolerance rather than assume adjacency. + +4. **A block can be decoded standalone** by wrapping its bytes in a synthetic + one-block chunk: copy the 32-byte header, set `nbytes` to the block's + uncompressed size, keep `blocksize`, set `cbytes = 36 + len(payload)`, append + `bstarts = [36]` and the payload. Verified for every block of chunks written + with zstd+shuffle/split, lz4+nosplit, zstd9+bitshuffle, and incompressible + data. + +5. **A chunk missing blocks is representable, and valid.** The format says a + stream with `csize == 0` is "fully made of zeros, and there is no cdata + section". So a placeholder block is `b"\0\0\0\0" * nstreams`, where + `nstreams` is `1` when the no-split flag is set and `typesize` otherwise — + four to thirty-two bytes, independent of codec and filters. Splicing header + + rewritten bstarts + (fetched blocks and placeholders) produces a chunk that + `blosc2.decompress2()` decodes with the fetched blocks exact and the rest + zeros, and that `SChunk.update_chunk()` accepts into a cache of the same + geometry. Verified across the same four configurations; the spliced chunk for + one block of the random-data case was 71 KB against 6.7 MB. + +6. **Block *k* is the k-th block of the chunk's flat buffer** (`k * blocksize`), + and sits at the C-order position `k` in the `ceil(chunks/blocks)` block grid. + Verified on a (120,100) array with chunks (60,50) and blocks (20,25). + +7. **memcpyed chunks (`clevel=0`, or incompressible at low clevel) have no + bstarts**: block *k* is raw at `32 + k * blocksize`. Byte ranges are trivial, + but a *spliced* memcpyed chunk is not expressible (there are no streams to + zero out), so those chunks need either a full-chunk fetch or a local + re-compression. + +8. **Special chunks** (zeros / NaN / uninit / repeated value) have no blocks and + no bytes in the file; `FsspecNDSource._special_chunk` (:930) already handles + them. + +9. **c-blosc2 already reads blocks one at a time over the io plugin.** For a + lazy chunk, `blosc2.c` :1804-1855 resolves the block's csize from the lazy + trailer and calls `io_cb->read(..., io_pos, fp)` for exactly that block, where + `io_cb` comes from `blosc2_get_io_cb(schunk->storage->io->id)` — i.e. from a + *pluggable* callback set (`blosc2_register_io_cb`, blosc2.h:1069). This is the + basis of route B below and contradicts the "lazy chunks cannot point at an + object store" reading of the format. + +## Route A — block fetching in Python (no C changes) + +### A.1 `FsspecNDSource` grows a block layer + +- `_chunk_layout(nchunk)`: one range read of `32 + 4 * nblocks` bytes at the + chunk offset (nblocks is not known before the header is read, so either read a + fixed optimistic prefix — the b2nd metalayer already gives `chunks` and + `blocks`, hence `nblocks`, so the size *is* known up front, one read) → + `(flags, blocksize, cbytes, bstarts, extents)` with `extents` from the + sorted-neighbour rule. Memoize per chunk on the source: the frame is immutable + between opens and `stamp` already detects replacement. +- `get_block(nchunk, nblock) -> bytes`, `aget_block` likewise: one `cat_file` + with the exact range. Stateless, so thread-safe like `get_chunk` is today. +- `get_blocks(nchunk, nblocks: list)`: sort by offset, coalesce runs separated by + less than some gap (a few KB), one request per run; keep the incidentally + fetched blocks rather than discarding them. +- The `MAX_OVERHEAD` cap and truncate-to-`cbytes` dance in `get_chunk` can stay: + the whole-chunk path remains the fallback for memcpyed chunks and for chunks + where most blocks are wanted. + +### A.2 Slice → blocks + +`blosc2.get_slice_nchunks` gives chunks; there is no block equivalent. It is +~15 lines of numpy given fact 6: for each touched chunk, intersect the slice with +the chunk's box, divide by `blocks` per dimension, take the C-order product of +the per-dimension block ranges. Start in Python; the array is small (blocks per +chunk, not per array). + +### A.3 Where partial chunks live — three options + +**A.3a — splice into the cache (recommended).** On fetch, read the cached chunk +(`_schunk_cache.get_chunk`, local and cheap), splice the newly fetched block +payloads in place of their zero-stream placeholders, rewrite bstarts, one +`update_chunk`. No compression, no decompression, anywhere on the path. Partial +progress persists for free, so a session that only ever touches part of a chunk +never re-downloads those blocks — including across runs, since the cache is a +real file. + +The objection is that a full-chunk read of the cache (`proxy[:]` without a +preceding fetch, or someone opening the cache file directly) sees zeros for +unfetched blocks. That hazard already exists at chunk granularity: an unfetched +chunk in the cache is a special/uninit chunk that reads as zeros, and +`LazyExpr._save` persists `Proxy._cache`'s urlpath and reopens it as a plain +NDArray ([src/blosc2/lazyexpr.py](../src/blosc2/lazyexpr.py):4735, 5217). +Blocks make the granularity finer, not the failure mode new. The bitmap in +`vlmeta` stays authoritative and every path that goes through `Proxy` fetches +first. + +**A.3b — buffer blocks in memory, write the chunk when complete.** The cache +only ever sees complete chunks, so nothing downstream can observe a hole. Costs: +a `dict[(nchunk, nblock), bytes]` that grows without bound for chunks that are +never completed (the common case for this feature — if the workload completed +chunks, block granularity would not be worth having), and no persistence of +partial progress, so every run re-downloads the same partial chunks. This +trades the feature's main benefit for a hazard that A.3a mostly already has. + +**A.3c — decompress and write through `cache[slice] = data`.** No format +surgery at all, works for memcpyed chunks too, but recompresses the whole chunk +on every partial write, and the write must be trimmed to `shape` at array edges. +Keep as the memcpyed fallback if a full-chunk fetch there is judged too coarse. + +### A.4 `Proxy` changes + +- `_fetched` becomes per block: `nchunks * blocks_per_chunk` bits, under a new + vlmeta key. `_load_fetched`'s legacy fallback stays (a `proxy-fetched` bitmap + from an older cache marks all blocks of its fetched chunks). At 1M chunks × 64 + blocks the bitmap is 8 MB in vlmeta; if that ever bites, store per-chunk + "complete" bits plus per-block bits only for incomplete chunks. Not now. +- `_missing_chunks` → `_missing_blocks(item)`, returning `(nchunk, [nblock])`. +- `_get_chunks` → `_get_blocks`: **two phases**, both fanned out over the same + thread pool — all chunk layouts first, then all blocks. This is what keeps the + extra round trip from multiplying by the number of chunks touched. +- `fetch`'s `finally: self._save_fetched()` pattern carries over unchanged. +- `afetch`/`aget_chunk` get the same treatment; only the sync path is exercised + by `memory://` tests, as today. + +### A.5 Sizing + +~120 lines in `FsspecNDSource`, ~120 in `Proxy`, ~20 for the slice→blocks +helper, ~150 of tests (request counting, partial progress within and across +runs, memcpyed / nblocks==1 / short last block / dict / special chunks, bitmap +migration). Roughly the size of phase 3 itself. + +## Route B — a Python io callback, block granularity for free + +Register a `blosc2_io_cb` whose `open`/`read`/`size` serve fsspec ranges, and +open the remote frame as an ordinary schunk through +`blosc2_schunk_open_offset_udio` (already called at +[src/blosc2/blosc2_ext.pyx](../src/blosc2/blosc2_ext.pyx):1747, 3406, 3422 for +the mmap and locking backends). Then fact 9 does the work: the C layer reads +lazy chunks and pulls exactly the blocks a getitem touches. This is route 3b of +the original plan, and it is the one that covers *every* container — sparse +frames, `.b2d` stores, plain SChunks, `offset != 0` — with no format parsing in +Python and no `Proxy` changes at all. + +What it costs, honestly: + +- **The GIL.** Blosc calls `io_cb->read` from its decompression threads. A + Python callback must acquire the GIL there. python-blosc2 already calls into + Python from those threads (prefilters/postfilters), so it is not unprecedented, + but a per-block network read serialized behind the GIL is a different traffic + profile than a prefilter. fsspec releases the GIL inside socket I/O, so + overlap is possible, but this needs prototyping before it is believed. +- **No batching, no prefetch.** Each block is a separate synchronous GET issued + from inside the decompression loop. Route A can coalesce ranges and fan out; + route B cannot without a read-ahead layer in the callback. +- **Request amplification on open.** The C frame reader does many small reads + (header, trailer, offsets, per-chunk headers). Each becomes a request unless + the callback wraps an fsspec caching file object (`blockcache`/`readahead`), + which is the obvious mitigation and is also where most of route B's simplicity + quietly goes. +- **`id` is a `uint8_t`** while the header's `BLOSC2_IO_USER_DEFINED` is 256, so + a registered id has to live in `[160, 255]` — worth confirming with upstream + before burning one. +- Cython work (~200 lines), so every edit means a full rebuild, and errors + surface as segfaults rather than tracebacks. + +Route B is the architecturally right answer and the one that survives a format +change. It is also the one that cannot be prototyped in an afternoon. + +## When does any of this actually pay? — measured + +Everything below is measured, not modelled: +[bench/ndarray/fsspec-block-granularity.py](../bench/ndarray/fsspec-block-granularity.py) +computes the touch ratios locally from an array's own chunk headers and then +replays both request patterns against a real object store. + +### The endpoint + +This machine to S3 `us-east-1`, anonymous public bucket, s3fs 2026.7.0: + +| | | +|---|---| +| one small range GET, serial | 226–248 ms | +| 8 small range GETs, pool of 8 | 280 ms total — a wave costs about one round trip | +| single-stream throughput | 3–5 MB/s | +| 8-stream aggregate throughput | ~12 MB/s | + +The 240 ms is transatlantic; in-region it would be 10–20 ms with far more +bandwidth. That moves both terms of the trade in the same direction, so the +break-even below is more portable than the individual numbers. + +### Touch ratios, three real arrays + +Bytes a slice needs in block mode (headers included) over bytes it needs in +chunk mode. `lung_raw_slice` is CT data (chunks 1.06 MB compressed, 32 +blocks/chunk), `tip_10` a benchmark table (13.5 MB, 125 blocks/chunk), `fancy` +a highly compressible ramp (0.12 MB, 250 blocks/chunk). + +| array | slice | chunks | blocks | chunk mode | block mode | ratio | +|---|---|---|---|---|---|---| +| lung | point | 1 | 1/32 | 1 req, 1.06 MB | 2 req, 0.03 MB | **3.2%** | +| lung | 32² window | 1 | 2/32 | 1 req, 1.06 MB | 3 req, 0.07 MB | **6.4%** | +| lung | one row | 6 | 43/192 | 6 req, 6.07 MB | 15 req, 1.52 MB | 25% | +| lung | one column | 10 | 39/320 | 10 req, 9.94 MB | 49 req, 1.23 MB | 12% | +| lung | one z-plane | 60 | 1677/1920 | 60 req, 57.3 MB | 120 req, 57.3 MB | 100% | +| tip_10 | point | 1 | 1/125 | 1 req, 13.5 MB | 2 req, 0.11 MB | **0.8%** | +| tip_10 | 1000 rows | 1 | 7/125 | 1 req, 13.5 MB | 3 req, 0.75 MB | **5.6%** | +| tip_10 | 50k rows | 3 | 313/375 | 3 req, 40.4 MB | 7 req, 33.7 MB | 83% | +| tip_10 | one column | 20 | 2500/2500 | 20 req, 269 MB | 40 req, 269 MB | 100% | +| fancy | point | 1 | 1/250 | 1 req, 0.12 MB | 2 req, 0.003 MB | 1.3% | +| fancy | 1M elements | 1 | 63/250 | 1 req, 0.12 MB | 2 req, 0.03 MB | 29% | + +### Wall time, replayed against real S3 + +Median of 5 interleaved repetitions, `max_concurrency=8`. "cached" is the same +fetch once the chunk headers have been read (the second and later slices of an +array, if the offsets are kept): + +| array | slice | chunk mode | blocks | blocks, cached | +|---|---|---|---|---| +| tip_10 | point | 3.77 s | 0.33 s **11x** | 0.15 s **26x** | +| tip_10 | one row | 5.77 s | 0.35 s **17x** | 0.15 s **39x** | +| tip_10 | 1000 rows | 5.25 s | 0.99 s **5.3x** | 0.26 s **20x** | +| tip_10 | 50k rows | 5.62 s | 5.11 s 1.1x | 3.50 s 1.6x | +| lung | 32² window | 0.79 s | 0.29 s **2.7x** | 0.15 s **5.4x** | +| lung | point | 0.49 s | 0.32 s 1.5x | 0.15 s 3.2x | +| lung | one row | 1.32 s | 0.94 s 1.4x | 0.29 s 4.5x | +| lung | one column | 0.63 s | 1.05 s **0.6x** | 0.76 s 0.8x | +| lung | half the array | 1.97 s | 2.82 s **0.7x** | 2.41 s 0.8x | +| fancy | point | 0.14 s | 0.29 s **0.5x** | 0.15 s 1.0x | +| fancy | 1M elements | 0.15 s | 0.30 s **0.5x** | 0.15 s 1.0x | + +### What the numbers say + +- **The win is real and large where it exists**: 5–17x on an array with 13 MB + chunks, 2.7x on one with 1 MB chunks. Not a marginal optimization. +- **Part of that win is parallelism, not bytes.** A slice touching one chunk is + *one* request in chunk mode, so it gets one TCP stream and 3–5 MB/s; block + mode splits it into several ranges that the pool runs at ~12 MB/s aggregate. +- **The loss is real too, and bounded**: 0.5–0.7x, i.e. exactly the one extra + wave, whenever the chunk is small (`fancy`, 0.12 MB) or the slice wants most + of its blocks anyway. +- **Break-even is ~0.5–1.5 MB of compressed chunk.** One extra wave costs ~0.15 s + here and a single stream moves ~3.5 MB/s, so the saving has to exceed ~0.5 MB; + in-region (15 ms, ~90 MB/s) the same arithmetic gives ~1.3 MB. The figure + barely moves with the endpoint, which makes it a usable constant. +- **Default block shapes are full in the trailing dimensions.** Every geometry + above has `blocks[-1] == chunks[-1]`, so selectivity exists only along the + leading dimensions: a *column* touches 100% of the blocks of every chunk it + touches, and block mode can only add requests (lung column: 49 requests + against 10, for a 0.6x). This is not an edge case, it is half of all slicing + patterns, so the whole-chunk fallback below is mandatory rather than an + optimization. +- **Request count matters as much as byte count** at 240 ms per wave. Coalescing + near-adjacent block ranges (4 KB gap tolerance) is what keeps lung's 43 blocks + down to 9 requests. + +Caveat on method: no writable bucket was available here, so the replay issues +the same request shape (count, sizes, phases, concurrency) against a 315 MB +public object rather than against an uploaded array. Transport, client stack and +latency are real; the bytes returned are not the array's. A run against a +genuine uploaded `.b2nd` would confirm the same numbers and cost a bucket. + +The extra round trip is the whole story, and there are four ways to spend less +of it, in increasing order of effort: + +1. **Batch the layout reads** (A.4): with N chunks touched and a pool of 8, the + cost is 2 round trips total, not 2N. This alone flips most multi-chunk slices. +2. **Persist the layouts.** bstarts is `4 * nblocks` bytes per chunk; keeping it + in the cache's vlmeta makes every later session one round trip per slice, and + the frame's `stamp` already invalidates it correctly. +3. **A whole-chunk threshold.** When the wanted blocks are more than about half + the chunk (by bytes, which the layout read gives exactly), fetch the chunk in + one request instead. This also covers memcpyed chunks and `nblocks == 1` for + free. +4. **Speculative layout read.** `nblocks` is known from the b2nd metalayer before + any request, so the layout read has an exact size; overlapping it with the + previous slice's block reads is possible but probably not worth it. + +The memory reduction (8 × block instead of 8 × chunk at peak) is unconditional +and may end up mattering more than the bytes. + +## Recommendation + +The measurement that gated this is done, and it says build it. + +1. **Build route A with A.3a**, plus mitigations 1-3 above. It is pure Python, it + composes with everything phase 3 already does (`max_concurrency`, the + persistent cache, the `stamp`), and the format work is verified rather than + speculative. Expected: 5–17x on multi-MB chunks, 2–5x on 1 MB chunks, and a + bounded 0.5x loss on everything else — which mitigation 3 turns into a wash. +2. **Mitigation 3 (the whole-chunk threshold) is not optional.** Half of all + slicing patterns touch every block of the chunks they touch, because default + block shapes are full in the trailing dimensions. The layout read gives the + exact wanted-bytes figure, so the rule is a one-liner: fetch the whole chunk + when the wanted blocks exceed ~50% of `cbytes`, or when `cbytes` is below the + ~1 MB break-even, or when the chunk is memcpyed. Everything else goes by block. +3. **Mitigation 2 (persist the layouts) is worth as much as the feature itself**: + it is another 2–3x on top (the "blocks, cached" column), for `4 * nblocks` + bytes per chunk in the cache's vlmeta. +4. **Keep route B as the answer for the formats route A cannot reach** (sparse + frames, `.b2d`, plain SChunks, `offset`), and prototype the GIL behaviour + before committing to it. Do not build both at once. +5. Do not touch `C2Array` — it has its own chunk endpoint and no block concept. + The new source methods must be optional (`getattr(src, "get_block", None)`), + with `Proxy` falling back to chunk granularity for sources that lack them. +6. Re-run the benchmark against a genuine uploaded `.b2nd` once a writable + bucket is at hand, to close the one methodological gap in the numbers above. + +## Adjacent bug found while measuring — fixed + +`SChunk.get_lazychunk()` returned only the 32-byte header for an ordinary chunk +of a file-backed frame, throwing away the `bstarts` and trailer sections that +make a lazy chunk useful — including the trailer's exact per-block compressed +sizes. The cap in `blosc2_ext.pyx` tested `chunk[31] & 0x70` (the special-value +bits) where the lazy flag is `0x08`, so "is this a real lazy chunk" was false for +every regular chunk and the buffer was truncated to `MAX_OVERHEAD`: + +```python +a = blosc2.open("lung_raw_slice.b2nd") # 1.05 MB chunks, 32 blocks each +len(a.schunk.get_lazychunk(0)) # was 32, now 300 = 32 + 32*4 + 12 + 32*4 +``` + +The `0x70` test was itself a workaround: testing `0x08` alone truncated the +repeated value off special chunks, which is what `iterchunks_info` reads. The cap +now applies only when the chunk is neither lazy nor special, so both work, and it +still keeps a whole in-memory chunk from being copied. Covered by +`test_get_lazychunk_sections` in `tests/test_schunk.py`, which pins the section +layout and the identity `header + bstarts + sum(block csizes) == cbytes`. + +Nothing had noticed because every caller (`iterchunks_info`, `batch_array`, +`objectarray`, `lazyexpr`) reads only header fields, and the sparse-gather path +calls `blosc2_schunk_get_lazychunk` from C without going through this wrapper. +It is not on the critical path for either route above — a byte-range reader +cannot call it — but it is what lets the benchmark read block offsets without +pulling whole chunks. + +## Note on `fsspec-blocks-ds4pro.md` + +An earlier analysis of the same question sits untracked at the repo root. Two of +its load-bearing claims do not survive contact with a real chunk: + +- "block *i* spans `[bstarts[i], bstarts[i+1])`" — false whenever the frame was + written with `nthreads > 1`, which is the default (fact 3). Block extents need + the sorted-neighbour rule. +- Its rejected alternative, "storing patched partial chunks in the container … + would silently serve garbage" — the missing blocks are not garbage but + format-defined zero streams (fact 5), and the alternative it prefers instead + (buffer until complete) is the one that throws away partial progress. That is + the trade this document flips. + +## Reproducing the format checks + +```python +import struct, numpy as np, blosc2 + +a = blosc2.arange(0, 2000 * 2000, dtype="f8", shape=(2000, 2000)) +chunk = a.schunk.get_chunk(0) +nbytes, cbytes, blocksize = blosc2.get_cbuffer_sizes(chunk) +nblocks = (nbytes + blocksize - 1) // blocksize +bstarts = np.frombuffer(chunk[32 : 32 + 4 * nblocks], dtype=" ImportError: Install s3fs to access S3 +gcs://bucket/key.b2nd -> ImportError: Please install gcsfs to access Google Storage +gs://bucket/key.b2nd -> ImportError: Please install gcsfs to access Google Storage +nosuchproto://b/k -> ValueError: Protocol not known: nosuchproto +``` + +The wording is not consistent between backends ("Install s3fs" vs "Please +install gcsfs"), which is one more reason to let fsspec own these strings +rather than mirroring them in our docs or asserting on them in tests. + +The second case matters because the branch fires on any `"://"`: a typo'd or +unsupported scheme produces a clear `ValueError` rather than falling through to +a misleading `FileNotFoundError`. Neither case needs handling from us; both +should be covered by a negative test in tier 1. + +`fsspec` itself is imported lazily, inside the branch, so a missing extra costs +an `ImportError` rather than an import-time cost for everybody. + +**Read.** One branch in `blosc2.open()` +([src/blosc2/schunk.py](src/blosc2/schunk.py):2075, +immediately after the `pathlib.PurePath` normalisation and before the +`.b2d`/`.b2z`/`.b2e` dispatch): + +```python +if "://" in urlpath and not urlpath.startswith(("file://", "http://", "https://")): + if mode != "r": + raise NotImplementedError("remote URLs can only be opened with mode='r'") + import fsspec + + with fsspec.open(urlpath, "rb") as f: + return blosc2.from_cframe(f.read()) +``` + +Notes on the details: + +- The `file://` exclusion lets fsspec-style local URLs keep working through + the normal local path, which supports mmap and every container format. + *(This turned out to need more than the exclusion: nothing downstream stripped + the scheme, so a `file://` URL was taken as a literal filename and failed. It + is normalized to a native path now, in `open()`, `NDArray.save()`, `Storage` + and the two constructor paths that bypass `Storage`.)* +- `offset != 0` should raise for now; the embedded-object case is a phase-3 + concern. +- `copy=False` on `from_cframe` is tempting (it pins the read buffer instead of + copying it) but the buffer is a throwaway `bytes` we just built, so `copy=True` + and `copy=False` cost the same peak memory here and `False` merely keeps the + buffer alive longer. Leave the default. + +**Write.** The mirror, in the save helpers rather than in `open()`: +`blosc2.save_array` / `save_tensor` +([src/blosc2/core.py](src/blosc2/core.py):528, +750) grow the same URL test and become +`fsspec.open(urlpath, "wb").write(arr.to_cframe())`. `NDArray.copy(urlpath=...)` +and friends keep rejecting remote URLs — the C layer writes incrementally and +cannot target an object store. + +**Documented limits of phase 1**, stated in the docstring rather than +discovered by users: + +- the whole object is read into memory, twice at the peak (the `bytes` plus the + reconstructed container) unless `copy=False`; +- read-only; +- single-file formats only (`.b2nd`, `.b2f`, `.b2e`, `.b2z`); `.b2d` and sparse + frames raise a clear `NotImplementedError` naming phase 2. + +## Phase 2 — Local cache, full format coverage — DONE + +The lazy way to get every container format, mmap, and repeat-run speed without +writing a byte-range reader. + +**As implemented:** `blosc2.open(url, cache_storage=...)`, backed by +`localize_fsspec_url()` in +[src/blosc2/core.py](src/blosc2/core.py), which +returns a local path that `open()` then re-enters with. All four open questions +below were settled as recommended, plus these decisions taken while building it: + +- **One knob, not two.** `cache_storage=` alone turns caching on; there is no + separate `cache=True`, since an explicit directory already says everything a + boolean would. No module-level `set_remote_cache()` either — add it if someone + asks. +- **`check_files=True` is mandatory, and was not free.** fsspec's `filecache` + does *not* check staleness by default (`check_files=False`), contrary to what + this plan assumed: it served a cached array whose remote bytes had changed. + Caught by a test that mutates the object between two opens. `simplecache` was + not added as a flag; it is what fsspec's own default already behaved like, and + nobody has asked for it. +- **Directory containers carry their own manifest.** fsspec has no `filecache` + equivalent for a prefix, so `.b2d` stores and sparse frames are fetched with + one `fs.get(recursive=True)` into a URL-hashed subdirectory, alongside a JSON + manifest of the remote `fs.find(detail=True)` listing. A changed listing + re-fetches the whole prefix; no per-file delta sync. +- **Unset kwargs are not a request.** The no-cache path rejects `mmap_mode`, + `offset` and friends by pointing at `cache_storage=`, but ignores kwargs whose + value is `None` — `load_tensor()` passes `dparams=None` unconditionally. +- Tests grew to 20, still `memory://` only: cache hit (no refetch), staleness + re-fetch for both files and directories, mmap over a cached file, a `.b2d` + `DictStore`, and a sparse frame. + +Write-back stayed out, as the section below says it should. + +The rest of this section is the original design. + +fsspec's `filecache` downloads an object once into a local cache directory and +hands back a real local file path. `blosc2.open()` on that path is the ordinary +local path, so *everything* works: sparse frames, `.b2d` directories (via +`fs.get()` of the prefix), `mmap_mode`, `offset`. Verified working against +`memory://` in scoping. + +Shape of it: + +```python +def _localize(urlpath, cache_storage=None): + """Download a remote container into the local fsspec cache, return its path.""" +``` + +- single-file containers: `fsspec.filesystem("filecache", target_protocol=..., + cache_storage=...).open(key).name`; +- directory containers (`.b2d`, sparse frames): `fs.get(prefix, localdir, + recursive=True)` into the same cache root, return the local directory. + +Open questions to settle before implementing: + +- **Cache location and lifetime.** Default to `platformdirs`-style user cache + or require an explicit `cache_storage=`? An unbounded implicit cache that + silently fills a laptop disk is the classic footgun here; an explicit + argument is the honest default, with a module-level + `blosc2.set_remote_cache(...)` for people who want it global. +- **Staleness.** `filecache` checks the remote mtime; S3 ETags make that + cheap-ish but not free (one HEAD per open). `simplecache` skips the check + entirely. Probably: `filecache` by default, `simplecache` behind a flag. +- **Interaction with phase 1.** Once phase 2 exists, phase 1's in-memory read + is still the right default for a one-shot read of a small object. Suggested + rule: `blosc2.open(url)` stays in-memory; `blosc2.open(url, cache=True)` (or a + global setting) goes through the cache. Do not silently switch behaviour. + +Phase 2 also unlocks write-back for single-file containers — write locally, +`fs.put()` on close — but that is a separate, opt-in `mode="w"` story and +should not be smuggled in with the read work. + +## Phase 3 — Byte-range chunk access — DONE, via 3a + +**As implemented:** `blosc2.open(url, lazy=True)` returns a `Proxy` over the new +`blosc2.FsspecNDSource` +([src/blosc2/proxy.py](src/blosc2/proxy.py)), +which reads the frame's header and offsets at open (three small reads) and then +one range read per chunk a slice touches. Measured on a 36 KB frame over +`memory://`: 276 bytes at open, 2 KB for a 50-element slice. + +Route 3a was chosen over the plan's recommendation of prototyping 3b first, +because **the offset-table blocker turned out to be much smaller than this plan +assumed**: + +- The frame header *is* a msgpack array. `msgpack.unpackb(header)` yields + `header_len`, the compressed size, the chunk size and the metalayer map with no + byte arithmetic at all. Exactly one field has to be located by hand — + `header_len`, at byte 0x0B — because it is needed to know how much to unpack. +- The `b2nd` metalayer then gives shape, chunks, blocks and dtype, so no sparse + local skeleton or C accessor is needed to describe the array. +- The offsets are one Blosc2 chunk at `header_len + compressed_size`, decompressed + with `blosc2.decompress2`. Two corrections to what this plan and the format doc + say: the offsets are relative to the **end of the header**, not to its + beginning; and a *negative* offset is not a position but a run-length chunk + (zeros, NaN, uninitialized) that was never written, which the source rebuilds + locally. + +That is about 45 lines of format knowledge, against 3b's Cython callback bridge, +its permanent registry id and its per-block `open()`. The judgement stands that +3b is the architecturally cleaner one — if the frame format ever grows a variant +this parser does not know, it will be 3b that survives it — but at this size 3a +was not worth deferring for it. Everything the parser reads is validated by the +tests decompressing real chunks through it. + +`aget_chunk` is implemented too, since it is the reason the plan kept 3a on the +table: `Proxy.afetch` overlaps up to 8 chunk fetches on async backends (s3fs and +friends), and falls back to the blocking path elsewhere. Only the fallback is +covered by tests — `memory://` is not async — so the concurrent path is the one +piece of this work that a real S3 endpoint would exercise first. + +Later, the batching the plan wanted from `aget_chunk` was given to the *sync* +path as well, since that is the one ordinary slicing uses: `get_chunk` became a +single stateless range read (it cost two, one for the chunk header), which made +it thread-safe, and `Proxy.fetch` grew a `max_concurrency=` thread pool. +Threads rather than asyncio, because driving `afetch` from `__getitem__` would +mean `asyncio.run()` inside a sync method — a `RuntimeError` in any notebook, +and the first sync-over-async in the library. The test asserts overlap with a +barrier rather than a stopwatch, since `memory://` has no latency to hide. + +It defaults to 8 rather than to serial. The *cost of being wrong* is small and +measured: over `memory://`, where the pool can only lose, a 100-chunk read goes +from 1.1 ms to 2.2 ms, about 10 µs per chunk. The gain is 7.4x on a 100-chunk +read against a 5 ms simulated round trip +([examples/ndarray/concurrent-fsspec.py](examples/ndarray/concurrent-fsspec.py)), +and unmeasured against a real endpoint. That asymmetry, plus `afetch` already +defaulting to 8 for remote sources, made serial-by-default the inconsistent +choice rather than the conservative one. + +That example had to invent its own latency: nothing that runs offline has any. +`memory://`, `zip://` and `tar://` are local reads, the regime where the pool +only costs, and `http://` is reserved for Caterva2 so a local server is not +reachable either. It subclasses fsspec's in-memory filesystem with a fixed delay +and says so, rather than implying a benchmark it cannot run. + +Two examples cover the feature: +[rw-fsspec.py](examples/ndarray/rw-fsspec.py) +for the three read modes and the write, and `concurrent-fsspec.py` for +`max_concurrency`. + +`lazy=True` and `cache_storage=` compose rather than excluding each other, which +is a departure from how phase 2 framed the choice: `cache_storage` means "where +this container's local copy lives", and `lazy` decides whether that copy is the +whole thing or only the chunks touched so far. A persistent chunk cache is +stamped with `fs.ukey()` and thrown away when that changes, since chunks fetched +by offsets from a replaced frame are not merely stale but wrong. The stamp +started as a hand-rolled `[size, mtime or LastModified]` tuple and was wrong: +see the testing note below. + +Not done: `lazy=True` needs a contiguous frame carrying a `b2nd` metalayer. +Plain SChunks, sparse frames and `.b2d` stores raise and point at +`cache_storage=`. `offset != 0` likewise raises. + +The rest of this section is the original design, including 3b, which stays +unbuilt. + +Only worth doing when someone actually has a container too large to download +and wants to slice a small part of it. Two candidate designs. They are *not* +strictly ranked: 3b is correct and complete, 3a is the one that can be fast. + +### 3a — `ProxyNDSource` over byte ranges + +Implement the +[src/blosc2/proxy.py](src/blosc2/proxy.py):38 +interface with `get_chunk(nchunk)` doing `fs.read_block(url, offset, length)`, +mirroring what `C2Array.get_chunk` does over HTTP +([src/blosc2/c2array.py](src/blosc2/c2array.py):372). +The `Proxy` machinery then caches decompressed chunks locally, and the async +`aget_chunk` hook can prefetch several ranges at once — which, per the latency +discussion below, is the whole reason this design stays on the table. + +The blocker: this needs the frame's **chunk offset table**, and python-blosc2 +exposes no way to get chunk offsets out of a cframe without opening it first. +So 3a requires either a small pure-Python frame-header/trailer parser (fragile, +duplicates format knowledge that belongs in C) or a new C-level accessor. Cost +this honestly before choosing it. + +### 3b — A user-defined I/O plugin bridging to an fsspec file object + +Register a `blosc2_io_cb` whose `open`/`read`/`size` callbacks reach an fsspec +file object, and open through `blosc2_schunk_open_offset_udio`. The C library +then does its own range reads, no format knowledge leaks into Python, and +`offset != 0`, sparse frames and every container format are fixed at once. + +**This needs no c-blosc2 modification to be correct.** Two facts establish that: + +- `frame.c` routes every read through + `blosc2_get_io_cb(frame->schunk->storage->io->id)` — around 148 call sites, + covering the header, the offsets table, chunk fetch and lazy-block fetch. + Nothing bypasses the callback table. +- The python-blosc2 side is already plumbed: `blosc2_schunk_open_offset_udio` + is called at + [src/blosc2/blosc2_ext.pyx](src/blosc2/blosc2_ext.pyx):1747, + 3406 and 3422. A new backend only has to supply the `blosc2_io{id, name, + params}` struct. + +Three constraints to design around, in increasing order of how much they hurt: + +**Registration id must be ≥ 160.** `blosc2_register_io_cb` rejects any id below +`BLOSC2_IO_REGISTERED` (blosc2.c:6813), and `id` is a `uint8_t`, so the usable +range is 160–255 — the Blosc plugin-registry range. (`BLOSC2_IO_USER_DEFINED = +256` is unreachable through a `uint8_t`.) So the id has to be coordinated with +upstream rather than picked freely. Registration is also process-global and +permanent: `g_ios` is a fixed array with no unregister call. + +**`open()` is called per block, not per file.** frame.c:163-167 restricts the +cached-handle path to `BLOSC2_IO_FILESYSTEM`: + +```c +if (io->id != BLOSC2_IO_FILESYSTEM) { + // Third-party backends keep the documented one-handle-per-reader contract + return io_cb->open(frame->urlpath, "rb", io->params); +} +``` + +Every other backend re-opens on each `frame_reader_acquire()`, and `blosc_d` +acquires once per *lazy block* (blosc2.c:1806-1854). The mmap backend survives +this only because its `open` is a cheap pointer return. So the fsspec backend +must be equally cheap: the `params` struct holds an already-open handle and +`open()` just hands the pointer back, doing no Python work — no refcounting, no +GIL — at open/close time. Workable, but it constrains the design from the start. + +**`read` runs concurrently on the blosc worker threads.** `blosc_d` is the +per-block decompression function, so any Python touched inside `read` has to +take the GIL and block reads serialise against each other. Probably acceptable +— S3 round-trip latency dominates, and the GIL is released while fsspec waits +on the socket — but measure it under `nthreads > 1` rather than assuming. +Prototype against a local file first, where the GIL cost is visible without +network noise masking it. + +### Where a c-blosc2 change would actually pay + +Not required, but worth proposing upstream if 3b goes ahead: + +- **Handle caching for third-party ids.** Either extend the reader cache past + `BLOSC2_IO_FILESYSTEM`, or add a flag on `blosc2_io_cb` + (`caches_handles` / `open_is_cheap`) that lets a backend opt in. Small and + localised to `frame_reader_acquire()`; removes the second constraint above. + +- **A prefetch/range-coalescing hook.** This is the limitation no flag fixes: + `blosc2_io_cb` has no way to express *"I will need blocks X through Y"*, so + c-blosc2 issues one range GET per block with no batching. Against local disk + that is fine; against S3 it is pure latency, one round trip per block. There + is no cheap upstream fix — it would mean a scatter/gather read callback or a + readahead hint in the I/O API — which is exactly why 3a's `aget_chunk` batching + keeps its appeal despite the offset-table problem. + +### Recommendation + +Do not start phase 3 speculatively; wait for a concrete user with a container +too big for phase 2's cache. When that arrives, prototype 3b first — it is the +architecturally correct one and needs no upstream change to work — and measure +against a real S3 endpoint before deciding whether the per-block round trips +justify the extra machinery of 3a. + +*Superseded: 3a shipped instead, and without waiting for the user. See the notes +at the top of this section — the offset table cost 45 lines of msgpack reading +rather than the C accessor this plan feared, which changed the arithmetic. 3b +stays unbuilt and its analysis below stays valid.* + +## Testing + +The point here is that **almost none of this needs AWS, credentials, or a new +dependency**. + +**Tier 1 — `memory://`, always on, no extra deps.** The dispatch and +serialisation path is protocol-generic, so fsspec's built-in in-memory +filesystem covers it. Lives in `tests/test_fsspec.py`, no marker, runs in the +default suite: + +```python +def test_fsspec_roundtrip(): + a = blosc2.arange(10, dtype="i4") + with fsspec.open("memory://x.b2nd", "wb") as f: + f.write(a.to_cframe()) + assert np.array_equal(blosc2.open("memory://x.b2nd")[:], a[:]) +``` + +Plus negative tests: `mode="a"` raises, `offset != 0` raises, `http://` still +routes to `C2Array`, `.b2d` raises the phase-2 `NotImplementedError`. These +catch every regression that is actually about *our* code. + +Phase 2's cache path is equally testable this way — `filecache` over +`target_protocol="memory"` with `cache_storage=tmp_path` gives a local file and +a real cache-hit assertion (open twice, assert one remote fetch), again with no +network. + +Skip condition: `pytest.importorskip("fsspec")`, since fsspec is optional. + +**Tier 2 — real S3, opt-in.** One test against a public anonymous bucket, +marked `network`. [pytest.ini](pytest.ini) +already excludes that marker from the default run +(`-m "not network and not heavy and not tui"`), so CI stays offline and the +test is run deliberately: + +``` +conda run -n blosc2 pytest -m network tests/test_s3.py +``` + +Requires `s3fs`, already in the `dev` group. It needs a stable, publicly +readable object to point at — either one published under a Blosc-controlled +bucket as part of this work, or a well-known open dataset. Decide which before +writing the test; do not let it depend on a bucket that can vanish. + +**Tier 3 — a real S3 protocol, locally.** `moto[server]` gives a local +S3-compatible endpoint that `s3fs` can be pointed at with `endpoint_url`. This +is a new dev dependency and it is only worth adding once phase 3 exists, since +`memory://` cannot exercise range requests and `moto` can. Not before. + +*Decided after phase 3 shipped: still no.* Three things `memory://` cannot +reach — the async `aget_chunk` path (memory is not an async backend), reads +through a buffered/block-caching file object, and the actual latency win — and +`moto` only buys the first two. That is not worth a dev dependency plus a local +HTTP server, and `blockcache::memory://` covers the second for free. Revisit if +the async path is ever to be locked down before someone runs this against real +S3. + +**What `memory://` gets wrong, and it is not what you would expect.** It is not +too *unrealistic* — it is **poorer in metadata than any real backend**, exposing +no `mtime` or `LastModified`. That silently degraded the hand-rolled cache stamp +to size-only, so a frame replaced by one of identical size was served from a +stale chunk cache; and the staleness test passed anyway, because compression had +made the two frames different sizes. `moto` would have *hidden* this bug, not +caught it. The fixes were `fs.ukey()`, which asks fsspec what identifies these +bytes rather than guessing which fields a backend exposes, and a test that +writes both frames uncompressed so only a real content check can pass. + +**What not to build:** no `boto3` stubbing, no fixture framework, no +per-protocol parametrisation across `s3`/`gcs`/`az`. The code path is one +branch; one filesystem exercising it is enough. + +## Recommendation + +**Ship phase 1 and stop.** It is roughly ten lines plus the extra plus the +tier-1 tests, it covers "my arrays are in S3 and I want to read them", and it is +the only phase whose value is certain today. + +*Superseded: phase 1 shipped, and phase 2 followed immediately after rather than +waiting for someone to hit the memory ceiling — it turned out to be about forty +lines and it closes the format gap (`.b2d`, sparse frames, mmap, offset), which +is worth more than the pause was. The "and stop" now applies at phase 3, where +the reasoning below is unchanged.* + +Note the asymmetry that argues for the pause: every open question below except +the last two is a *phase 2* question. The caching layer is where the design +decisions and the footguns live, not the feature itself. Wait until someone +actually hits phase 1's memory ceiling before paying for that. + +Phase 3 stays parked. It is the most interesting engineering here and the least +justified: it needs either a frame-offset parser we do not have (3a) or a C +callback bridge constrained by the GIL and per-block `open()` (3b), to serve a +user who has not shown up yet. + +*Superseded: 3a shipped. The frame-offset parser we "do not have" was 45 lines, +which is what changed the answer — not a user showing up.* + +## Open Questions, With Recommendations + +Each of these is stated in context in the phase it belongs to; this is the +summary and the current leaning. + +The four phase-2 questions are now **settled, each the way it was recommended**: +explicit `cache_storage=` with no default, staleness-checked on every open +(which took `check_files=True`, see the phase 2 notes), caching opt-in so +`open(url)` is unchanged, and no tier-2 network test. + +The two phase-3 questions are **moot**: 3a needs no I/O plugin, so no registry id +was burned and no upstream change is on the table. + +What replaces them is one open question, the only one left in this document: +**how the concurrent fetch behaves against a real endpoint.** Both paths that +overlap requests — the thread pool the sync path uses by default, and +`aget_chunk`'s async one, which `memory://` never even enters — are correct by +test and plausible by arithmetic, and neither has met real S3. The first thing +to do with a bucket in hand is check that 8 in flight is a sensible default +rather than one that trips throttling. + +- **Phase 2 cache location and lifetime.** *Recommendation: require an explicit + `cache_storage=`, no implicit default.* An implicit `platformdirs` cache that + silently fills a laptop disk with multi-GB arrays is the classic footgun, and + there is no eviction story we would want to write. Explicit is one argument + the caller types once. + +- **Phase 2 `filecache` vs `simplecache`.** *Recommendation: `filecache`, i.e. + pay the HEAD-per-open to check staleness.* Silently serving a stale array is + the worst failure mode this feature has; one round trip against S3 latency is + not the thing to optimise. `simplecache` behind a flag for callers who know + their data is immutable. + +- **Does phase 2 change what `open(url)` does?** *Recommendation: no.* + In-memory stays the default and caching is opt-in (`cache=True`, or a + module-level setting). Silently upgrading a working call from "one GET" to + "writes a file on your disk" is the kind of surprise that generates issues. + +- **Which bucket the tier-2 network test points at.** *Recommendation: do not + write tier 2 at first.* `memory://` covers every line of code we own; a + real-S3 test exercises fsspec and AWS, and buys a permanent dependency on an + object staying public. Add it the first time a bug escapes tier 1 — and if it + does, publish the fixture under a Blosc-controlled bucket rather than + borrowing someone's open dataset. + +- **The `blosc2_io_cb` id ≥ 160 (phase 3).** *Recommendation: ignore until + phase 3 is greenlit.* The constraint is "coordinate with the Blosc plugin + registry", which is not a real obstacle for this project — but do not burn an + id speculatively, since registration is permanent. + +- **Whether to propose the c-blosc2 handle-caching flag upstream (phase 3).** + *Recommendation: only after measuring, and only if phase 3 happens.* The + per-block `open()` is a real cost but a speculative one until a workload shows + it. + +One thing that is **not** an open question, recorded here so it does not become +one: `mode="a"` on a remote URL should raise permanently, not "for now". S3 has +no rename and no locks, so append semantics would be a trap rather than a +missing feature. + +## Rollout Notes + +- Phase 1 is additive: a URL that previously raised `FileNotFoundError` now + works. Nothing existing changes behaviour, so it does not need a deprecation + cycle. +- Document the extra in the install docs alongside `[tui]` / `[hires]`. Say + "any fsspec URL" rather than enumerating protocols — the list is fsspec's to + grow, not ours to track — and give `pip install "blosc2[fsspec]" s3fs` as the + S3 recipe so the backend install is visible rather than implied. +- State explicitly that credentials are configured through `s3fs`/AWS + conventions, not through blosc2. +- Keep the error message for a missing `fsspec` actionable: + `pip install "blosc2[fsspec]"`. Missing *backends* are fsspec's error to + raise, not ours to intercept. +- Chained URLs (`filecache::s3://bucket/key.b2nd`, + `zip://inner.b2nd::s3://bucket/archive.zip`) go through the same + `fsspec.open()` call and work for free, including reading a Blosc2 file + straight out of a remote zip. Worth one line in the docs; worth no code. diff --git a/pyproject.toml b/pyproject.toml index d4324c254..c7bffaefc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,6 +59,10 @@ 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"] [project.scripts] parquet-to-blosc2 = "blosc2.cli.parquet_to_blosc2:main" @@ -73,6 +77,7 @@ dev = [ "matplotlib", "pandas", "plotly", + "moto[server]", "pre-commit", "pyarrow", "ruff", @@ -82,6 +87,16 @@ dev = [ ] test = [ "pytest", + # tests/test_fsspec.py importorskips fsspec, so without this the whole fsspec + # feature silently skips in CI. memory:// covers the protocol-generic paths, + # and is also the configuration most users installing [fsspec] are in. + "fsspec; platform_machine != 'wasm32'", + # tests/test_fsspec_s3.py needs a real S3 endpoint (moto, served locally, so + # still offline) and a real *async* backend (s3fs). memory:// is neither, and + # cannot see the class of bug that lives there: awaiting an async filesystem's + # own coroutine fails on s3fs and passes silently on memory://. Runs in ~3 s. + "moto[server]; platform_machine != 'wasm32'", + "s3fs; platform_machine != 'wasm32'", # pytest.ini defaults to `-n auto`; where xdist is absent (wasm32, or a # bare `pip install pytest`) the root conftest.py degrades it to a serial run "pytest-xdist; platform_machine != 'wasm32'", diff --git a/src/blosc2/__init__.py b/src/blosc2/__init__.py index 764f8f78b..648014b58 100644 --- a/src/blosc2/__init__.py +++ b/src/blosc2/__init__.py @@ -588,7 +588,16 @@ def _raise(exc): result_type, can_cast, ) -from .proxy import Proxy, ProxySource, ProxyNDSource, ProxyNDField, SimpleProxy, jit, as_simpleproxy +from .proxy import ( + Proxy, + ProxySource, + ProxyNDSource, + ProxyNDField, + FsspecNDSource, + SimpleProxy, + jit, + as_simpleproxy, +) from .indexing import Index from .schunk import SChunk, load, open @@ -870,6 +879,7 @@ def _raise(exc): "NDArray", "NDField", "Operand", + "FsspecNDSource", "Proxy", "ProxyNDField", "ProxyNDSource", diff --git a/src/blosc2/blosc2_ext.pyx b/src/blosc2/blosc2_ext.pyx index ca13dac6c..4b257c686 100644 --- a/src/blosc2/blosc2_ext.pyx +++ b/src/blosc2/blosc2_ext.pyx @@ -2067,12 +2067,17 @@ cdef class SChunk: cbytes = blosc2_schunk_get_lazychunk(self.schunk, nchunk, &chunk, &needs_free) if cbytes < 0: raise RuntimeError("Error while getting the lazychunk") - # The next does not always work (bug) - # cdef uint8_t is_lazy = chunk[BLOSC2_MAX_OVERHEAD - 1] & 0x08 - # Workaround - cdef uint8_t is_lazy = chunk[BLOSC2_MAX_OVERHEAD - 1] & 0x70 - if not is_lazy: - # Put a cap on the buffer size for the non-lazy chunk + if chunk == NULL: + raise RuntimeError(f"Chunk {nchunk} holds no data") + # Two kinds of chunk carry something past the header that the caller needs: + # a lazy one (0x08), whose bstarts and trailer say where its blocks are on + # disk, and a special one (0x70), whose repeated value follows the header. + # Anything else is a whole chunk sitting in memory, and copying it here + # would defeat the point of asking for a lazy one: cap it at the header. + # Testing only 0x70, as this did for a while, capped every lazy chunk of a + # file-backed frame down to its header and threw the block offsets away. + cdef uint8_t blosc2_flags = chunk[BLOSC2_MAX_OVERHEAD - 1] + if not (blosc2_flags & (0x08 | 0x70)): cbytes = MAX_OVERHEAD ret_chunk = PyBytes_FromStringAndSize(chunk, cbytes) if needs_free: diff --git a/src/blosc2/core.py b/src/blosc2/core.py index 524982302..6b36db274 100644 --- a/src/blosc2/core.py +++ b/src/blosc2/core.py @@ -11,16 +11,21 @@ import copy import ctypes import ctypes.util +import hashlib import json import math import os import pathlib import pickle import platform +import re +import shutil import subprocess import sys +import urllib.parse +import urllib.request from dataclasses import asdict -from functools import lru_cache +from functools import cache, lru_cache from typing import TYPE_CHECKING, ClassVar import numpy as np @@ -534,7 +539,9 @@ def save_array(arr: np.ndarray, urlpath: str, chunksize: int | None = None, **kw The NumPy array to be saved. urlpath: str - The path for the file where the array will be saved. + The path for the file where the array will be saved. An fsspec URL + (``s3://``, ``gs://``, ``memory://``...) writes the whole container in one + shot; it needs the ``fsspec`` extra and the protocol driver installed. chunksize: int The size (in bytes) for the chunks during compression. If not provided, @@ -612,6 +619,132 @@ def load_array(urlpath: str, dparams: dict | None = None) -> np.ndarray: return load_tensor(urlpath, dparams=dparams) +def normalize_urlpath(urlpath: object) -> object: + """Turn a `file://` URL into the native path it names, leaving anything else alone. + + Local URLs are kept off the fsspec branch so they can use mmap and every + container format, which only works if the scheme is stripped first. + """ + if isinstance(urlpath, str) and urlpath.startswith("file://"): + parsed = urllib.parse.urlparse(urlpath) + netloc = "" if parsed.netloc.lower() in ("", "localhost") else parsed.netloc + if re.fullmatch("[A-Za-z]:", netloc): + if os.name != "nt": + raise ValueError( + f"{urlpath} names the host {netloc!r}; only Windows can reach one, as a drive" + ) + # A Windows drive lands in netloc for the two-slash form, `file://C:/x` + prefix = netloc + elif not netloc: + prefix = "" + elif os.name == "nt": + # A real authority is a UNC host, which keeps its two slashes + prefix = "//" + netloc + else: + raise ValueError( + f"{urlpath} names the host {netloc!r}; only Windows can reach one, as a UNC path" + ) + return urllib.request.url2pathname(prefix + parsed.path) + return urlpath + + +def is_fsspec_url(urlpath: object) -> bool: + """Whether *urlpath* should be routed through fsspec. + + Any URL with a scheme qualifies, except `file://` (which the local path + handles better, with mmap and every container format) and `http(s)://` + (reserved for :ref:`C2Array`). Chained URLs such as + `zip://x.b2nd::s3://bucket/a.zip` qualify too, as fsspec resolves them. + """ + return ( + isinstance(urlpath, str) + and "://" in urlpath + and not urlpath.startswith(("file://", "http://", "https://")) + ) + + +def _import_fsspec(urlpath: str): + """Import fsspec with an actionable error when the extra is not installed.""" + try: + import fsspec + except ImportError: + raise ImportError( + f'Reading or writing {urlpath} requires fsspec: pip install "blosc2[fsspec]"' + ) from None + # Missing protocol backends (s3fs, gcsfs...) are fsspec's error to raise. + return fsspec + + +def fsspec_open(urlpath: str, mode: str): + """`fsspec.open()`, but complaining properly when fsspec is missing.""" + return _import_fsspec(urlpath).open(urlpath, mode) + + +def fsspec_cache_path(urlpath: str, cache_storage: str | pathlib.Path, suffix: str = "") -> str: + """The local path under *cache_storage* reserved for *urlpath*, creating the directory.""" + os.makedirs(cache_storage, exist_ok=True) + name = hashlib.sha256(urlpath.encode()).hexdigest() + return os.path.join(str(cache_storage), name + suffix) + + +@cache +def _suffixed_cache_mapper(): + """fsspec's cache naming, plus the extension `blosc2.open()` dispatches on. + + A `.b2e` store is told apart from a bare SChunk by its name alone, so a cached + copy under fsspec's plain hash would silently open as the wrong type. + """ + from fsspec.implementations.cache_mapper import AbstractCacheMapper + + class SuffixedCacheMapper(AbstractCacheMapper): + def __call__(self, path: str) -> str: + return hashlib.sha256(path.encode()).hexdigest() + pathlib.PurePosixPath(path).suffix + + return SuffixedCacheMapper() + + +def localize_fsspec_url(urlpath: str, cache_storage: str | pathlib.Path) -> str: + """Materialize the container at *urlpath* under *cache_storage*, return its local path. + + Single-file containers go through fsspec's ``filecache``, which downloads + once and afterwards pays one HEAD per open to check staleness. Directory + containers (``.b2d`` stores, sparse frames) have no such layer in fsspec, so + the whole prefix is fetched and re-fetched whenever the remote listing stops + matching the manifest written at download time. + """ + fsspec = _import_fsspec(urlpath) + from fsspec.utils import tokenize + + cache_storage = str(cache_storage) + fs, path = fsspec.url_to_fs(urlpath) + + if not fs.isdir(path): + # check_files is off by default in fsspec, which would happily serve a + # cached copy of an array that changed remotely -- the worst failure mode + # this feature has, and worth one HEAD per open to avoid. + opts = { + "cache_storage": cache_storage, + "check_files": True, + "cache_mapper": _suffixed_cache_mapper(), + } + with fsspec.open(f"filecache::{urlpath}", "rb", filecache=opts) as f: + return f.name + + localdir = fsspec_cache_path(urlpath, cache_storage) + manifest = pathlib.Path(localdir + ".json") + # tokenize(info) is what fs.ukey() hashes, so this asks each backend what + # identifies a file rather than guessing which fields it exposes -- size and + # mtime miss a same-size rewrite, and memory:// has no mtime at all + listing = json.dumps( + {name: tokenize(entry) for name, entry in sorted(fs.find(path, detail=True).items())} + ) + if not manifest.exists() or manifest.read_text() != listing: + shutil.rmtree(localdir, ignore_errors=True) + fs.get(path.rstrip("/") + "/", localdir, recursive=True) + manifest.write_text(listing) + return localdir + + def pack_tensor( tensor: tensorflow.Tensor | torch.Tensor | np.ndarray, chunksize: int | None = None, **kwargs: dict ) -> bytes | int: @@ -656,6 +789,14 @@ def pack_tensor( """ arr = np.asarray(tensor) + # Object stores cannot be written incrementally, so build the whole cframe in + # memory and PUT it in one go. + remote_urlpath = kwargs.get("urlpath") if is_fsspec_url(kwargs.get("urlpath")) else None + if remote_urlpath is not None: + del kwargs["urlpath"] + # A remote write always replaces, but reading mode still forbids one + blosc2_ext.check_access_mode(remote_urlpath, kwargs.pop("mode", "a")) + schunk = blosc2.SChunk(chunksize=chunksize, data=arr, **kwargs) # Guess the kind of tensor / array @@ -674,6 +815,12 @@ def pack_tensor( schunk.vlmeta["__pack_tensor__"] = (kind, arr.shape, dtype) + if remote_urlpath is not None: + cframe = schunk.to_cframe() + with fsspec_open(remote_urlpath, "wb") as f: + f.write(cframe) + return len(cframe) + if schunk.urlpath is None: return schunk.to_cframe() else: @@ -762,7 +909,9 @@ def save_tensor( The tensor or array to be saved. urlpath: str - The file path where the tensor or array will be saved. + The file path where the tensor or array will be saved. An fsspec URL + (``s3://``, ``gs://``, ``memory://``...) writes the whole container in one + shot; it needs the ``fsspec`` extra and the protocol driver installed. chunksize: int The size (in bytes) for the chunks during compression. If not provided, diff --git a/src/blosc2/indexing.py b/src/blosc2/indexing.py index f244641de..21e044af7 100644 --- a/src/blosc2/indexing.py +++ b/src/blosc2/indexing.py @@ -153,10 +153,10 @@ def _cleanup_in_memory_store(key: int) -> None: _IN_MEMORY_INDEXES.pop(key, None) _IN_MEMORY_INDEX_FINALIZERS.pop(key, None) scope = ("memory", key) - stale_data = [cache_key for cache_key in _DATA_CACHE if cache_key[0] == scope] + stale_data = [cache_key for cache_key in tuple(_DATA_CACHE) if cache_key[0] == scope] for cache_key in stale_data: _DATA_CACHE.pop(cache_key, None) - stale_handles = [cache_key for cache_key in _SIDECAR_HANDLE_CACHE if cache_key[0] == scope] + stale_handles = [cache_key for cache_key in tuple(_SIDECAR_HANDLE_CACHE) if cache_key[0] == scope] for cache_key in stale_handles: _SIDECAR_HANDLE_CACHE.pop(cache_key, None) _hot_cache_clear(scope=("memory", key)) @@ -755,10 +755,10 @@ def _hot_cache_clear(scope: tuple[str, str | int] | None = None) -> None: """Clear all in-process hot cache entries for *scope* (or all scopes).""" global _HOT_CACHE_BYTES if scope is not None: - keys = [key for key in _HOT_CACHE if key[0] == scope] + keys = [key for key in tuple(_HOT_CACHE) if key[0] == scope] for key in keys: _HOT_CACHE_BYTES -= _HOT_CACHE.pop(key).nbytes - _HOT_CACHE_ORDER[:] = [key for key in _HOT_CACHE_ORDER if key[0] != scope] + _HOT_CACHE_ORDER[:] = [key for key in list(_HOT_CACHE_ORDER) if key[0] != scope] return _HOT_CACHE.clear() _HOT_CACHE_ORDER.clear() @@ -1068,10 +1068,10 @@ def _data_cache_key(array: blosc2.NDArray, token: str, category: str, name: str) def _clear_cached_data(array: blosc2.NDArray, token: str) -> None: prefix = (_array_key(array), token) - keys = [key for key in _DATA_CACHE if key[:2] == prefix] + keys = [key for key in tuple(_DATA_CACHE) if key[:2] == prefix] for key in keys: _DATA_CACHE.pop(key, None) - handle_keys = [key for key in _SIDECAR_HANDLE_CACHE if key[:2] == prefix] + handle_keys = [key for key in tuple(_SIDECAR_HANDLE_CACHE) if key[:2] == prefix] for key in handle_keys: _SIDECAR_HANDLE_CACHE.pop(key, None) @@ -1093,7 +1093,7 @@ def _invalidate_sidecar_cache_entries(array: blosc2.NDArray, token: str, categor for cache_category in categories: _DATA_CACHE.pop(_data_cache_key(array, token, cache_category, name), None) prefix = _sidecar_handle_cache_key(array, token, cache_category, name) - for key in [k for k in _SIDECAR_HANDLE_CACHE if k[:4] == prefix[:4]]: + for key in [k for k in tuple(_SIDECAR_HANDLE_CACHE) if k[:4] == prefix[:4]]: _SIDECAR_HANDLE_CACHE.pop(key, None) diff --git a/src/blosc2/ndarray.py b/src/blosc2/ndarray.py index 211f251bd..914bc5f81 100644 --- a/src/blosc2/ndarray.py +++ b/src/blosc2/ndarray.py @@ -32,6 +32,7 @@ import blosc2 from blosc2 import SpecialValue, blosc2_ext, compute_chunks_blocks +from blosc2.core import fsspec_open, is_fsspec_url, normalize_urlpath from blosc2.info import InfoReporter, format_nbytes_info from blosc2.schunk import SChunk @@ -315,6 +316,7 @@ def are_partitions_behaved(shape, chunks, blocks): bool True if the partitions are well-behaved, False otherwise. """ + def check_contiguity(container, part): if container and container[-1] != part[-1]: return False @@ -5046,9 +5048,14 @@ def save(self, urlpath: str, contiguous=True, **kwargs: Any) -> None: Parameters ---------- urlpath: str - The path where the array will be saved. + The path where the array will be saved. An fsspec URL (needing the + ``fsspec`` extra) uploads the whole array as a single object, which + replaces whatever was there: object stores have no partial write, so + this is the only shape a remote write can take, and two writers to + the same key silently lose one. contiguous: bool, optional - Whether to save the array contiguously. + Whether to save the array contiguously. A sparse frame is a directory + and so cannot be saved to an fsspec URL. Other Parameters ---------------- @@ -5071,6 +5078,20 @@ def save(self, urlpath: str, contiguous=True, **kwargs: Any) -> None: >>> # Save the array to a file >>> a.save("array.b2frame") """ + urlpath = normalize_urlpath(urlpath) + if is_fsspec_url(urlpath): + if not contiguous: + raise NotImplementedError( + "a sparse frame is a directory, so it cannot be saved to an fsspec URL" + ) + # An object store takes the whole thing at once, and always replaces, + # but reading mode still forbids a write + blosc2_ext.check_access_mode(urlpath, kwargs.pop("mode", "w")) + array = self.copy(**kwargs) if kwargs else self + with fsspec_open(urlpath, "wb") as f: + f.write(array.to_cframe()) + return + blosc2_ext.check_access_mode(urlpath, "w") # Add urlpath to kwargs kwargs["urlpath"] = urlpath @@ -6746,7 +6767,8 @@ def save(array: NDArray, urlpath: str, contiguous=True, **kwargs: Any) -> None: array: :ref:`NDArray` The array to be saved. urlpath: str - The path to the file where the array will be saved. + The path to the file where the array will be saved, or an fsspec URL to + upload it to as a single object. See :meth:`NDArray.save`. contiguous: bool, optional Whether to store the array contiguously. @@ -7017,6 +7039,11 @@ def _check_ndarray_kwargs(**kwargs): # noqa: C901 # If a key appears in both operands, the one from the right-hand operand wins kwargs = storage_dflts | kwargs + if kwargs.get("urlpath") is not None: + # A Storage instance normalized (and vetted) its own on construction; a bare + # kwarg or a `storage=` mapping never reached __post_init__, so borrow it here + kwargs["urlpath"] = blosc2.Storage(urlpath=kwargs["urlpath"]).urlpath + supported_keys = [ "chunks", "blocks", diff --git a/src/blosc2/proxy.py b/src/blosc2/proxy.py index 425416359..fd5c5f9be 100644 --- a/src/blosc2/proxy.py +++ b/src/blosc2/proxy.py @@ -8,9 +8,12 @@ import ast import asyncio import inspect +import os +import struct import textwrap from abc import ABC, abstractmethod from collections.abc import Sequence +from concurrent.futures import ThreadPoolExecutor try: from numpy.typing import DTypeLike @@ -225,6 +228,18 @@ def __init__( mode: str, optional "a" means read/write (create if it doesn't exist); "w" means create (overwrite if it exists). Default is "a". + + With "a" and an existing :paramref:`urlpath`, the cache written by an + earlier run is adopted as is, so whatever it already holds is not + fetched from the source again. It must be a cache from a proxy over a + source of the same geometry (shape, dtype and partitioning); anything + else raises rather than being silently reused or overwritten. + + Geometry is all that is checked unless the source can name the exact + bytes it reads, as :ref:`FsspecNDSource` does with its ``stamp``: a + source whose contents changed underneath while its geometry did not + is adopted, and the cache keeps serving what the earlier run fetched. + Pass ``mode="w"`` when the source may have been rewritten. kwargs: dict, optional Keyword arguments supported: @@ -247,12 +262,26 @@ def __init__( kwargs = {} self._cache = kwargs.pop("_cache", None) vlmeta = kwargs.pop("vlmeta", None) + caterva2_env = kwargs.pop("caterva2_env", False) + + 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 + # still in there, and the creation path below would refuse to build + # over an existing container anyway + if kwargs: + # The container already exists, so these would be quietly dropped + raise ValueError( + f"{', '.join(kwargs)} cannot be applied to the existing cache at {urlpath}; " + f"pass mode='w' to build it anew" + ) + self._cache = self._reopen_cache(urlpath) - if self._cache is None: + fresh = self._cache is None + if fresh: meta_val = { "local_abspath": None, "urlpath": None, - "caterva2_env": kwargs.pop("caterva2_env", False), + "caterva2_env": caterva2_env, } container = getattr(self.src, "schunk", self.src) if hasattr(container, "urlpath"): @@ -283,8 +312,19 @@ def __init__( ) self._cache.fill_special(self.src.nbytes // self.src.typesize, blosc2.SpecialValue.UNINIT) self._schunk_cache = getattr(self._cache, "schunk", self._cache) + # What is already cached cannot be read off the cache itself: a chunk that + # is a run of a single value (zeros, NaNs, whatever blosc2.full() writes) + # is stored as a special chunk once fetched, telling it apart from the + # empty ones the cache starts life with. Hence an explicit bitmap. + nchunks = self._schunk_cache.nchunks + self._fetched = bytearray((nchunks + 7) // 8) if fresh else self._load_fetched(nchunks) if self.urlpath is None: self.urlpath = getattr(self._schunk_cache, "urlpath", None) + # Geometry alone cannot tell a replaced source from the one the cache was + # filled from, so record whatever identity the source can name itself by + stamp = getattr(self.src, "stamp", None) + if stamp is not None: + self._schunk_cache.vlmeta["fsspec-stamp"] = stamp if vlmeta: for key in vlmeta: self._schunk_cache.vlmeta[key] = vlmeta[key] @@ -293,6 +333,91 @@ def __enter__(self) -> "Proxy": """Enter a context manager and return this proxy.""" return self + def _load_fetched(self, nchunks: int) -> bytearray: + """The bitmap of already fetched chunks that a previous run left behind.""" + stored = self._schunk_cache.vlmeta.get("proxy-fetched") + if stored is not None and len(stored) == (nchunks + 7) // 8: + return bytearray(stored) + # A cache filled before the bitmap existed, or one handed over through + # `_cache=`: everything that is not a special chunk was surely fetched + fetched = bytearray((nchunks + 7) // 8) + self._fetched = fetched + for info in self._schunk_cache.iterchunks_info(): + if info.special == blosc2.SpecialValue.NOT_SPECIAL: + self._mark_fetched(info.nchunk) + return fetched + + def _mark_fetched(self, nchunk: int) -> None: + self._fetched[nchunk // 8] |= 1 << (nchunk % 8) + + def _missing_chunks(self, item) -> list[int]: + """The chunks *item* touches, minus those already in the cache.""" + nchunks = self._schunk_cache.nchunks + # Full realization when item is (), else only the chunks it intersects + wanted = range(nchunks) if item == () else list(blosc2.get_slice_nchunks(self._cache, item)) + return [int(n) for n in wanted if not self._fetched[n // 8] >> (n % 8) & 1] + + def _save_fetched(self) -> None: + """Persist the bitmap, so a later run does not fetch these chunks again. + + Called even when a fetch failed partway: whatever did arrive is kept. + """ + self._schunk_cache.vlmeta["proxy-fetched"] = bytes(self._fetched) + + def _reopen_cache(self, urlpath: str): + """Adopt the cache container stored at *urlpath*, checking it fits the source.""" + from blosc2.schunk import _set_default_dparams + + # Not blosc2.open(): that would rebuild the source we already hold, and + # raise outright for sources it cannot reconstruct from the cache metadata + kwargs = {} + _set_default_dparams(kwargs) + cached = blosc2.blosc2_ext.open(str(urlpath), "a", 0, **kwargs) + schunk = getattr(cached, "schunk", cached) + if "proxy-source" not in schunk.meta: + raise ValueError( + f"{urlpath} is not a proxy cache; pass mode='w' to overwrite it or choose another urlpath" + ) + # Chunk *numbers* are the currency between cache and source, so the + # partitioning has to match, not just the logical shape: fetch() would + # otherwise ask the source for chunk n meaning something else entirely + # A cache of the other kind is a mismatch in itself, and asking it for a + # shape it does not have would raise AttributeError instead of saying so + if hasattr(self.src, "shape") != hasattr(cached, "shape"): + raise ValueError( + f"the cache at {urlpath} is a {type(cached).__name__}, which does not fit a " + f"{type(self.src).__name__} source" + ) + if hasattr(self.src, "shape"): + fields = "shape, dtype, chunks, blocks" + here = (tuple(cached.shape), cached.dtype, tuple(cached.chunks), tuple(cached.blocks)) + there = ( + tuple(self.src.shape), + np.dtype(self.src.dtype), + tuple(self.src.chunks), + tuple(self.src.blocks), + ) + else: + fields = "nbytes, chunksize, typesize" + here = (schunk.nbytes, schunk.chunksize, schunk.typesize) + there = (self.src.nbytes, self.src.chunksize, self.src.typesize) + if here != there: + raise ValueError( + f"the cache at {urlpath} was built for a different source: it holds {here}, " + f"the source is {there} ({fields})" + ) + # Same geometry is not the same bytes: a replaced remote frame keeps its + # layout while every cached chunk, and every offset it was fetched by, + # goes stale. Only for sources that can name themselves; the rest are + # adopted on geometry alone, as documented. + stamp = getattr(self.src, "stamp", None) + if stamp is not None and schunk.vlmeta.get("fsspec-stamp") != stamp: + raise ValueError( + f"the cache at {urlpath} was built against different remote bytes; " + f"pass mode='w' to fetch them anew" + ) + return cached + def __exit__(self, exc_type, exc_val, exc_tb) -> bool: """Exit a context manager. @@ -301,7 +426,9 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> bool: """ return False - def fetch(self, item: slice | list[slice] | None = ()) -> blosc2.NDArray | blosc2.schunk.SChunk: + def fetch( + self, item: slice | list[slice] | None = (), max_concurrency: int | None = None + ) -> blosc2.NDArray | blosc2.schunk.SChunk: """ Get the container used as cache with the requested data updated. @@ -310,6 +437,12 @@ def fetch(self, item: slice | list[slice] | None = ()) -> blosc2.NDArray | blosc item: slice or list of slices, optional If not None, only the chunks that intersect with the slices in items will be retrieved if they have not been already. + max_concurrency: int, optional + Maximum number of `get_chunk` calls to run at once, in a thread + pool. Only worth raising for sources whose fetches are dominated by + round-trip latency, and only safe for sources whose `get_chunk` is + thread-safe, such as :ref:`FsspecNDSource`. Defaults to the source's + own `max_concurrency` attribute if it has one, else 1 (serial). Returns ------- @@ -329,22 +462,35 @@ def fetch(self, item: slice | list[slice] | None = ()) -> blosc2.NDArray | blosc [2 3] [4 5]] """ - if item == (): - # Full realization - for info in self._schunk_cache.iterchunks_info(): - if info.special != blosc2.SpecialValue.NOT_SPECIAL: - chunk = self.src.get_chunk(info.nchunk) - self._schunk_cache.update_chunk(info.nchunk, chunk) - else: - # Get only a slice - nchunks = blosc2.get_slice_nchunks(self._cache, item) - for info in self._schunk_cache.iterchunks_info(): - if info.nchunk in nchunks and info.special != blosc2.SpecialValue.NOT_SPECIAL: - chunk = self.src.get_chunk(info.nchunk) - self._schunk_cache.update_chunk(info.nchunk, chunk) + missing = self._missing_chunks(item) + try: + for nchunk, chunk in self._get_chunks(missing, max_concurrency): + self._schunk_cache.update_chunk(nchunk, chunk) + self._mark_fetched(nchunk) + finally: + if missing: + self._save_fetched() return self._cache + def _get_chunks(self, nchunks: list[int], max_concurrency: int | None): + """Yield (nchunk, chunk) pairs, overlapping the fetches when asked to.""" + if max_concurrency is None: + max_concurrency = getattr(self.src, "max_concurrency", 1) + if max_concurrency <= 1 or len(nchunks) < 2: + for nchunk in nchunks: + yield nchunk, self.src.get_chunk(nchunk) + return + # Writing to the cache stays on this thread; only the fetches fan out + pool = ThreadPoolExecutor(max_workers=min(max_concurrency, len(nchunks))) + try: + yield from zip(nchunks, pool.map(self.src.get_chunk, nchunks), strict=True) + finally: + # map() queues every chunk up front, so without cancel_futures an error + # here (or a Ctrl-C) would first sit through thousands of pending + # requests; only the handful already running are waited for + pool.shutdown(cancel_futures=True) + async def afetch( self, item: slice | list[slice] | None = (), max_concurrency: int | None = None ) -> blosc2.NDArray | blosc2.schunk.SChunk: @@ -431,18 +577,14 @@ async def afetch( if not callable(getattr(self.src, "aget_chunk", None)): raise NotImplementedError("afetch is only available if the source has an aget_chunk method") - if item == (): - wanted = None # every missing chunk - else: - wanted = set(blosc2.get_slice_nchunks(self._cache, item)) - to_fetch = [ - info.nchunk - for info in self._schunk_cache.iterchunks_info() - if info.special != blosc2.SpecialValue.NOT_SPECIAL and (wanted is None or info.nchunk in wanted) - ] + to_fetch = self._missing_chunks(item) if max_concurrency is None: - max_concurrency = REMOTE_MAX_CONCURRENCY if isinstance(self.src, blosc2.C2Array) else 1 + max_concurrency = getattr( + self.src, + "max_concurrency", + REMOTE_MAX_CONCURRENCY if isinstance(self.src, blosc2.C2Array) else 1, + ) semaphore = asyncio.Semaphore(max(1, max_concurrency)) async def _fetch_one(nchunk): @@ -450,9 +592,13 @@ async def _fetch_one(nchunk): chunk = await self.src.aget_chunk(nchunk) # Runs to completion between awaits, so concurrent writers can't interleave. self._schunk_cache.update_chunk(nchunk, chunk) + self._mark_fetched(nchunk) if to_fetch: - await asyncio.gather(*(_fetch_one(nchunk) for nchunk in to_fetch)) + try: + await asyncio.gather(*(_fetch_one(nchunk) for nchunk in to_fetch)) + finally: + self._save_fetched() return self._cache @@ -587,6 +733,219 @@ def fields(self) -> dict: return {key: ProxyNDField(self, key) for key in _fields} +_FRAME_MAGIC = b"b2frame\0" + + +def _read_frame_index(f) -> tuple[bytes, list, np.ndarray]: + """Read the header and the chunk offsets of a contiguous frame from *f*. + + Returns the raw header bytes, the header decoded as the msgpack array it is, + and the absolute position of every chunk. A negative position is not a + position at all: it encodes a run-length chunk that was never written to the + file. See ``README_CFRAME_FORMAT.rst`` in c-blosc2 for the layout. + """ + import msgpack + + f.seek(0) + prefix = f.read(24) + if prefix[2:10] != _FRAME_MAGIC: + raise ValueError("not a Blosc2 contiguous frame") + # header_len is the one field that must be located by hand; everything after + # it comes out of unpacking the header, which is plain msgpack + header_len = struct.unpack(">i", prefix[11:15])[0] + f.seek(0) + raw = f.read(header_len) + # raw=True because the flags field is a msgpack *string* holding four raw + # bytes, and codec_flags packs clevel into its high nibble: from clevel 8 up + # that byte is not valid UTF-8 and decoding the header blows up + header = msgpack.unpackb(raw, raw=True, strict_map_key=False) + + # An empty frame has no chunks, so it has no offsets chunk either: what sits + # at index_pos is the trailer, and reading it as one fails obscurely + if header[8] == 0: # chunksize + return raw, header, np.empty(0, dtype=np.int64) + + # The offsets live in a Blosc2 chunk of their own, right after the data ones + index_pos = header[1] + header[5] + f.seek(index_pos) + index_cbytes = struct.unpack("= 0, offsets + header_len, offsets) + + +class _RangeReader: + """The seek/read pair `_read_frame_index` needs, served by exact range requests.""" + + def __init__(self, fs, path: str): + self._fs, self._path, self._pos = fs, path, 0 + + def seek(self, pos: int) -> None: + self._pos = pos + + def read(self, size: int) -> bytes: + data = self._fs.cat_file(self._path, start=self._pos, end=self._pos + size) + self._pos += len(data) + return data + + +def _frame_metalayer(raw: bytes, header: list, name: str): + """Decode the *name* metalayer out of an already-read frame header.""" + offset = header[13][1][name.encode()] # KeyError if there is no such metalayer + nbytes = struct.unpack(">I", raw[offset + 1 : offset + 5])[0] # msgpack bin32 + import msgpack + + return msgpack.unpackb(raw[offset + 5 : offset + 5 + nbytes], raw=False) + + +def _chunk_extents(offsets: np.ndarray, header: list) -> np.ndarray: + """How many bytes to read at each chunk offset to be sure of covering it. + + A chunk carries its own compressed size in its header, but asking for that + first would cost a second request per chunk. The next thing stored after a + chunk bounds it instead -- another chunk, or the offsets chunk -- capped by + what a chunk can possibly weigh, so a hole left by an updated chunk cannot + turn into an absurd read. The caller truncates to the real size. + """ + index_pos = header[1] + header[5] + bounds = np.sort(np.append(offsets[offsets >= 0], index_pos)) + extents = bounds[np.searchsorted(bounds, offsets, side="right")] - offsets + return np.minimum(extents, header[8] + blosc2.MAX_OVERHEAD) + + +class FsspecNDSource(ProxyNDSource): + """A :ref:`Proxy` source that serves the chunks of a remote Blosc2 frame. + + The frame stays where it is: only its header, its chunk offsets, and the + chunks a slice actually touches ever cross the network. 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``:: + + src = blosc2.FsspecNDSource("s3://bucket/big.b2nd") + a = blosc2.Proxy(src, urlpath="big-cache.b2nd", mode="a") + + Contiguous frames carrying a ``b2nd`` metalayer only, which is what + :func:`blosc2.asarray` and friends write to a single file. Sparse frames and + ``.b2d`` stores are directories; open those with ``cache_storage=``. + + Parameters + ---------- + urlpath: str + The fsspec URL of the frame. + max_concurrency: int, optional + How many chunk fetches the enclosing :ref:`Proxy` may run at once. Each + chunk costs one range request, so against an object store a slice is + almost entirely round-trip latency and overlapping the requests is the + whole win. Defaults to 8, the same figure :meth:`Proxy.afetch` uses for + remote sources. Pass 1 for a protocol with no latency to hide, where + the thread pool costs about 10 microseconds per chunk and saves nothing. + """ + + def __init__(self, urlpath: str, max_concurrency: int = REMOTE_MAX_CONCURRENCY): + from blosc2.core import _import_fsspec + + self.max_concurrency = max_concurrency + fsspec = _import_fsspec(urlpath) + fs, path = fsspec.url_to_fs(urlpath) + if 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" + ) + self.urlpath = urlpath + self._fs, self._path = fs, path + # Identifies the remote bytes, so a cache built against them can tell it + # 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) + # Exact ranges, not fs.open(): a buffered handle reads a whole block per + # seek (50 MiB on s3fs by default), which would undo the point of a lazy + # open. Chunk reads are stateless, so nothing here is shared between threads + raw, header, self._offsets = _read_frame_index(_RangeReader(fs, path)) + self._chunksize = header[8] + self._extents = _chunk_extents(self._offsets, header) + try: + _, _, shape, chunks, blocks, dtype_format, dtype = _frame_metalayer(raw, header, "b2nd") + 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" + ) from None + if dtype_format != 0: + raise NotImplementedError(f"unsupported dtype format {dtype_format} in {urlpath}") + self._shape, self._chunks, self._blocks = tuple(shape), tuple(chunks), tuple(blocks) + try: + self._dtype = np.dtype(dtype) + except TypeError: + # Structured dtypes are stored as their repr, as blosc2_ext does too + self._dtype = np.dtype(ast.literal_eval(dtype)) + + @property + def shape(self) -> tuple: + return self._shape + + @property + def chunks(self) -> tuple: + return self._chunks + + @property + def blocks(self) -> tuple: + return self._blocks + + @property + def dtype(self) -> np.dtype: + return self._dtype + + def get_chunk(self, nchunk: int) -> bytes: + offset = int(self._offsets[nchunk]) + if offset < 0: + return self._special_chunk(offset) + data = self._fs.cat_file(self._path, start=offset, end=offset + int(self._extents[nchunk])) + return data[: struct.unpack(" bytes: + """Same as :meth:`get_chunk`, but without blocking the caller's event loop. + + This is what makes :meth:`Proxy.afetch` worth using against an object + store, where a slice spanning many chunks is nearly all round-trip + latency. + + The fetch goes to a worker thread rather than being awaited directly. + Awaiting an async filesystem's coroutine looks like the obvious thing to + do and does not work: fsspec drives those on a private event loop of its + own, so a client created there and awaited here raises "got Future + attached to a different loop" (seen with s3fs). Its blocking API is the + supported way in, and it hands off to that same private loop, so the + thread parks on a queue rather than on a socket. + """ + offset = int(self._offsets[nchunk]) + if offset < 0: + return self._special_chunk(offset) + loop = asyncio.get_running_loop() + return await loop.run_in_executor(None, self.get_chunk, nchunk) + + def _special_chunk(self, offset: int) -> bytes: + """Rebuild a run-length chunk, which lives in its offset instead of the file.""" + kind = ((offset & 0xFFFFFFFFFFFFFFFF) >> 56) & 0x7 + nitems = self._chunksize // self._dtype.itemsize + if kind == 2: + data = np.full(nitems, np.nan, dtype=self._dtype) + else: + # A run of zeros (1); uninitialized chunks (4) have no defined + # content, and zeros is what reading them locally hands back too + data = np.zeros(nitems, dtype=self._dtype) + # The blocksize has to be the container's: left to choose, blosc2 takes + # the whole chunk, and the cache then rejects the chunk we hand it + return blosc2.compress2( + data, + typesize=self._dtype.itemsize, + blocksize=int(np.prod(self._blocks)) * self._dtype.itemsize, + ) + + class ProxyNDField(blosc2.Operand): def __init__(self, proxy: Proxy, field: str): self.proxy = proxy diff --git a/src/blosc2/schunk.py b/src/blosc2/schunk.py index 4184e1432..6332f5f21 100644 --- a/src/blosc2/schunk.py +++ b/src/blosc2/schunk.py @@ -22,6 +22,13 @@ import blosc2 from blosc2 import SpecialValue, blosc2_ext +from blosc2.core import ( + fsspec_cache_path, + fsspec_open, + is_fsspec_url, + localize_fsspec_url, + normalize_urlpath, +) from blosc2.info import InfoReporter, format_nbytes_info from blosc2.msgpack_utils import msgpack_packb, msgpack_unpackb @@ -366,7 +373,14 @@ def __init__( # noqa: C901 if isinstance(kwargs.get("dparams"), blosc2.DParams): kwargs["dparams"] = asdict(kwargs.get("dparams")) - urlpath = kwargs.get("urlpath") + urlpath = normalize_urlpath(kwargs.get("urlpath")) + if urlpath is not None: + kwargs["urlpath"] = urlpath + if is_fsspec_url(urlpath): + raise ValueError( + f"{urlpath} is an fsspec URL, which cannot back a container as it is written; " + f"build it in memory and write to_cframe() there, or use NDArray.save()" + ) if "contiguous" not in kwargs: # Make contiguous true for disk, else sparse (for in-memory performance) kwargs["contiguous"] = urlpath is not None @@ -1930,6 +1944,90 @@ def _finalize_special_open(special, urlpath, mode): return special +def _lazy_fsspec_proxy( + urlpath: str, cache_storage: 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. + """ + # 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) + 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: + # The remote frame was replaced, which makes every cached chunk -- and + # every offset they were fetched by -- meaningless + 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") + + +def _cache_stamp(path: str): + """The remote stamp a cached proxy container was built against, if any. + + None for a cache that cannot be read at all, which an interrupted run can + leave behind: the caller throws those away just like a stale one. + """ + _set_default_dparams(kwargs := {}) + try: + cache = blosc2_ext.open(path, "r", 0, **kwargs) + except RuntimeError: + return None + return getattr(cache, "schunk", cache).vlmeta.get("fsspec-stamp") + + +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 + 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 + 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. + """ + 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) + max_concurrency = kwargs.pop("max_concurrency", None) + if kwargs.pop("lazy", False): + if offset != 0: + raise NotImplementedError("offset is not supported with lazy=True") + 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) + + 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 offset != 0: + raise NotImplementedError("offset on an fsspec URL requires passing cache_storage=") + # 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=") + 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" + ) + with fsspec_open(urlpath, "rb") as f: + return blosc2.from_cframe(f.read()) + + def open( urlpath: str | pathlib.Path | blosc2.URLPath, mode: str = "r", @@ -1956,7 +2054,9 @@ def open( ---------- urlpath: str | pathlib.Path | :ref:`URLPath` The path where the :ref:`SChunk` (or :ref:`NDArray`) - is stored. If it is a remote array, a :ref:`URLPath` must be passed. + is stored. If it is a remote Caterva2 array, a :ref:`URLPath` must be passed. + Any other URL with a scheme (``s3://``, ``gs://``, ``zip://``, ``memory://``...) + is opened through fsspec; see the `Notes` section for the limits. mode: str, optional Persistence mode: 'r' means read only (must exist); 'a' means read/write (create if it doesn't exist); @@ -1974,6 +2074,25 @@ def open( An offset in the file where super-chunk or array data is located (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 + where it is and reads the chunks a slice touches, one range request + each, instead of transferring the whole thing. Contiguous frames + holding an :ref:`NDArray` only. The fetched chunks are kept in memory, + or in ``cache_storage`` when that is given as well. + max_concurrency: int, optional + Only with ``lazy``: how many chunk fetches to run at once, in a + thread pool. A slice against an object store is almost entirely + round-trip latency, so overlapping the requests is what makes a wide + slice 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 + Only for fsspec URLs: a directory holding this container's local + copy — the whole thing, or just the chunks ``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. 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. @@ -2014,6 +2133,16 @@ def open( * If :paramref:`urlpath` is a :ref:`URLPath` instance, :paramref:`mode` must be 'r', :paramref:`offset` must be 0, and kwargs cannot be passed. + * fsspec URLs need the ``fsspec`` extra (``pip install "blosc2[fsspec]"``) and + the driver for the protocol (``s3fs``, ``gcsfs``...), which fsspec asks for + by name when it is missing; credentials are configured there, not here. + ``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 + in its own way. + * Persistent data handling follows a strict no-hidden-writes rule: - ``mode='r'`` is observational only and never mutates the opened object. @@ -2075,6 +2204,10 @@ def open( if isinstance(urlpath, pathlib.PurePath): urlpath = str(urlpath) + urlpath = normalize_urlpath(urlpath) + + if is_fsspec_url(urlpath): + return _open_fsspec_url(urlpath, mode, offset, kwargs) # Keep explicit store paths on the direct dispatch path. For regular # Blosc containers, try the standard open first and only fall back to the diff --git a/src/blosc2/storage.py b/src/blosc2/storage.py index c74c25278..fd22229ba 100644 --- a/src/blosc2/storage.py +++ b/src/blosc2/storage.py @@ -10,6 +10,7 @@ from dataclasses import asdict, dataclass, field, fields import blosc2 +from blosc2.core import is_fsspec_url, normalize_urlpath def default_nthreads(): @@ -248,6 +249,14 @@ class Storage: meta: dict = None def __post_init__(self): + self.urlpath = normalize_urlpath(self.urlpath) + if is_fsspec_url(self.urlpath): + # The C layer writes a container incrementally, rewriting its header + # and offsets as chunks land; an object store has no partial writes + raise ValueError( + f"{self.urlpath} is an fsspec URL, which cannot back a container as it is " + f"written; build it in memory and NDArray.save() it there in one go" + ) if self.contiguous is None: self.contiguous = self.urlpath is not None # Check for None values diff --git a/tests/ndarray/test_proxy.py b/tests/ndarray/test_proxy.py index 17719b5dd..e82badd47 100644 --- a/tests/ndarray/test_proxy.py +++ b/tests/ndarray/test_proxy.py @@ -131,6 +131,76 @@ def test_readonly_proxy_keeps_both_readonly(tmp_path): np.testing.assert_array_equal(readonly_ctx[:], data) +def test_reuse_cache_across_runs(tmp_path): + proxy_path = str(tmp_path / "proxy.b2nd") + data = np.arange(120, dtype=np.int32).reshape(12, 10) + source = blosc2.asarray(data, chunks=(4, 5), blocks=(2, 5)) + + proxy = blosc2.Proxy(source, urlpath=proxy_path, mode="a") + np.testing.assert_array_equal(proxy[0:4, 0:5], data[0:4, 0:5]) + del proxy + + # mode="a" over an existing cache picks up what the previous run fetched + proxy = blosc2.Proxy(source, urlpath=proxy_path, mode="a") + assert proxy._cache.schunk.urlpath == proxy_path + np.testing.assert_array_equal(proxy[:], data) + + +def test_reuse_cache_rejects_construction_kwargs(tmp_path): + # The container already exists, so contiguous= (and any other kwarg meant for + # the constructor) would be quietly dropped instead of doing anything + proxy_path = str(tmp_path / "proxy.b2nd") + source = blosc2.asarray(np.arange(120, dtype=np.int32).reshape(12, 10), chunks=(4, 5)) + + blosc2.Proxy(source, urlpath=proxy_path, mode="a") + with pytest.raises(ValueError, match="contiguous"): + blosc2.Proxy(source, urlpath=proxy_path, mode="a", contiguous=False) + + +def test_reuse_cache_rejects_other_kind(tmp_path): + proxy_path = str(tmp_path / "proxy.b2f") + + class Source(blosc2.ProxySource): + nbytes, chunksize, typesize = 1000, 100, 1 + + def get_chunk(self, nchunk): + raise NotImplementedError + + blosc2.Proxy(Source(), urlpath=proxy_path, mode="a") + with pytest.raises(ValueError, match="does not fit"): + blosc2.Proxy(blosc2.asarray(np.arange(1000, dtype=np.int32)), urlpath=proxy_path, mode="a") + + +def test_reuse_cache_rejects_foreign_container(tmp_path): + path = str(tmp_path / "plain.b2nd") + blosc2.arange(0, 120, dtype=np.int32, shape=(12, 10), urlpath=path, mode="w") + source = blosc2.asarray(np.arange(120, dtype=np.int32).reshape(12, 10)) + + with pytest.raises(ValueError, match="not a proxy cache"): + blosc2.Proxy(source, urlpath=path, mode="a") + + +@pytest.mark.parametrize( + "other", + [ + lambda data: blosc2.asarray(np.arange(50, dtype=np.float64)), + # Same shape and dtype, different partitioning: chunk numbers are what + # the proxy passes to the source, so this would silently fetch the + # wrong chunk or run off the end + lambda data: blosc2.asarray(data, chunks=(2, 5), blocks=(1, 5)), + ], + ids=["shape", "chunks"], +) +def test_reuse_cache_rejects_mismatched_source(tmp_path, other): + proxy_path = str(tmp_path / "proxy.b2nd") + data = np.arange(120, dtype=np.int32).reshape(12, 10) + source = blosc2.asarray(data, chunks=(4, 5), blocks=(2, 5)) + blosc2.Proxy(source, urlpath=proxy_path, mode="a").fetch() + + with pytest.raises(ValueError, match="different source"): + blosc2.Proxy(other(data), urlpath=proxy_path, mode="a") + + # Test the ProxyNDSources interface @pytest.mark.parametrize( ("shape", "chunks", "blocks"), diff --git a/tests/test_fsspec.py b/tests/test_fsspec.py new file mode 100644 index 000000000..b9a90c716 --- /dev/null +++ b/tests/test_fsspec.py @@ -0,0 +1,723 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# This source code is licensed under a BSD-style license (found in the +# LICENSE file in the root directory of this source tree) +####################################################################### + +import os +import pathlib +import threading + +import numpy as np +import pytest + +import blosc2 + +fsspec = pytest.importorskip("fsspec") + + +@pytest.fixture(autouse=True) +def clean_memory_fs(): + fsspec.filesystem("memory").store.clear() + + +def test_open_memory_url(): + a = blosc2.arange(10, dtype="i4") + with fsspec.open("memory://x.b2nd", "wb") as f: + f.write(a.to_cframe()) + + b = blosc2.open("memory://x.b2nd") + assert isinstance(b, blosc2.NDArray) + assert np.array_equal(b[:], a[:]) + + +def test_save_array_to_url(): + a = np.arange(100, dtype="f8").reshape(10, 10) + nbytes = blosc2.save_array(a, "memory://y.b2nd") + assert nbytes > 0 + assert np.array_equal(blosc2.load_array("memory://y.b2nd"), a) + + +def test_save_tensor_to_url(): + a = np.arange(50, dtype="f4") + blosc2.save_tensor(a, "memory://z.b2nd") + assert np.array_equal(blosc2.load_tensor("memory://z.b2nd"), a) + + +def test_save_ndarray_to_url(): + a = blosc2.arange(0, 100, dtype="i4", shape=(10, 10), chunks=(5, 10)) + a.save("memory://sv.b2nd") + b = blosc2.open("memory://sv.b2nd") + assert np.array_equal(b[:], a[:]) + assert b.chunks == a.chunks + + +def test_module_save_to_url(): + a = blosc2.arange(0, 50, dtype="f8") + blosc2.save(a, "memory://sv2.b2nd") + assert np.array_equal(blosc2.open("memory://sv2.b2nd")[:], a[:]) + + +def test_save_to_url_honours_cparams(): + a = blosc2.arange(0, 100, dtype="i4", shape=(10, 10), chunks=(5, 10)) + a.save("memory://sv3.b2nd", cparams=blosc2.CParams(codec=blosc2.Codec.LZ4)) + b = blosc2.open("memory://sv3.b2nd") + assert b.schunk.cparams.codec == blosc2.Codec.LZ4 + assert np.array_equal(b[:], a[:]) + + +def test_save_sparse_to_url(): + a = blosc2.arange(10, dtype="i4") + with pytest.raises(NotImplementedError, match="sparse frame"): + a.save("memory://sv4.b2nd", contiguous=False) + + +@pytest.mark.parametrize( + "make", + [ + lambda: blosc2.zeros((10,), urlpath="memory://c.b2nd", mode="w"), + lambda: blosc2.asarray(np.arange(10), urlpath="memory://c.b2nd", mode="w"), + lambda: blosc2.arange(10).copy(urlpath="memory://c.b2nd", mode="w"), + lambda: blosc2.SChunk(chunksize=100, urlpath="memory://c.b2f", mode="w"), + ], + ids=["zeros", "asarray", "copy", "schunk"], +) +def test_container_cannot_be_backed_by_url(make): + # These write incrementally through the C layer, which an object store + # cannot serve; the error has to say so rather than fail deep in C + with pytest.raises(ValueError, match="save"): + make() + + +def test_schunk_roundtrip(): + schunk = blosc2.SChunk(chunksize=1000) + schunk.append_data(np.arange(1000, dtype="u1")) + with fsspec.open("memory://s.b2f", "wb") as f: + f.write(schunk.to_cframe()) + + sc = blosc2.open("memory://s.b2f") + assert isinstance(sc, blosc2.SChunk) + assert sc.nbytes == schunk.nbytes + + +def test_chained_url(tmp_path): + # A container inside a local zip, reached through fsspec's chained syntax + import zipfile + + a = blosc2.arange(20, dtype="i2") + zippath = tmp_path / "archive.zip" + with zipfile.ZipFile(zippath, "w") as zf: + zf.writestr("inner.b2nd", a.to_cframe()) + + b = blosc2.open(f"zip://inner.b2nd::file://{zippath}") + assert np.array_equal(b[:], a[:]) + + +@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") + + +def test_offset_needs_cache(): + with pytest.raises(NotImplementedError, match="cache_storage"): + blosc2.open("memory://x.b2nd", offset=32) + + +def test_mmap_needs_cache(): + with pytest.raises(NotImplementedError, match="cache_storage"): + blosc2.open("memory://x.b2nd", mmap_mode="r") + + +def test_dir_container_needs_cache(): + with pytest.raises(NotImplementedError, match="cache_storage"): + blosc2.open("memory://store.b2d") + + +def test_dir_container_with_query_needs_cache(): + with pytest.raises(NotImplementedError, match="cache_storage"): + blosc2.open("memory://store.b2d?version=1") + + +def test_cached_open(tmp_path): + a = blosc2.arange(10, dtype="i4") + with fsspec.open("memory://c.b2nd", "wb") as f: + f.write(a.to_cframe()) + + b = blosc2.open("memory://c.b2nd", cache_storage=tmp_path) + assert np.array_equal(b[:], a[:]) + assert any(tmp_path.iterdir()) + + +def test_cached_open_is_local(tmp_path): + # The cached container is a real local file, so mmap works on it + a = blosc2.arange(10, dtype="i4") + 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") + assert np.array_equal(b[:], a[:]) + + +def test_cache_hit_avoids_refetch(tmp_path, monkeypatch): + a = blosc2.arange(10, dtype="i4") + with fsspec.open("memory://h.b2nd", "wb") as f: + f.write(a.to_cframe()) + + fetches = [] + memfs = type(fsspec.filesystem("memory")) + orig = memfs._open + monkeypatch.setattr( + 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) + assert len(fetches) == 1 + blosc2.open("memory://h.b2nd", cache_storage=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,) + + 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,) + + +def test_cached_dict_store(tmp_path): + # A .b2d store is a directory, so it only works through the cache + localstore = str(tmp_path / "local.b2d") + with blosc2.DictStore(localstore, mode="w") as dstore: + dstore["/a"] = blosc2.arange(10, dtype="i4") + 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: + assert sorted(dstore.keys()) == ["/a", "/b"] + assert np.array_equal(dstore["/a"][:], np.arange(10, dtype="i4")) + + +def test_cached_dir_refetches_when_remote_changes(tmp_path): + memfs = fsspec.filesystem("memory") + cache = tmp_path / "cache" + localstore = str(tmp_path / "d.b2d") + 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: + 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: + assert sorted(dstore.keys()) == ["/a", "/b"] + + +def test_cached_sparse_frame(tmp_path): + localpath = str(tmp_path / "sparse.b2nd") + 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") + assert np.array_equal(b[:], a[:]) + + +def _put(name, arr): + fsspec.filesystem("memory").pipe_file("/" + name, arr.to_cframe()) + return "memory://" + name + + +@pytest.mark.parametrize("chunks", [(100,), (37,)]) +def test_lazy_roundtrip(chunks): + a = blosc2.arange(0, 1000, dtype="i4", chunks=chunks, blocks=(11,)) + p = blosc2.open(_put("lazy.b2nd", a), lazy=True) + assert (p.shape, p.chunks, p.blocks, p.dtype) == (a.shape, a.chunks, a.blocks, a.dtype) + assert np.array_equal(p[:], a[:]) + + +def test_lazy_multidim(): + a = blosc2.arange(0, 10000, dtype="f4", shape=(100, 100), chunks=(10, 100)) + p = blosc2.open(_put("lazy2d.b2nd", a), lazy=True) + assert np.array_equal(p[3:7, 20:30], a[3:7, 20:30]) + + +@pytest.mark.parametrize( + "arr", + [ + blosc2.zeros((1000,), dtype="f8", chunks=(100,)), + blosc2.full((1000,), np.nan, dtype="f8", chunks=(100,)), + # blocks != chunks: a rebuilt run-length chunk must carry the container's + # blocksize, not whatever blosc2 picks when left to choose + blosc2.zeros((1000,), dtype="f8", chunks=(100,), blocks=(10,)), + blosc2.zeros((4_000_000,), dtype="f8", chunks=(1_000_000,)), + blosc2.uninit((1000,), dtype="i4", chunks=(100,), blocks=(10,)), + ], + ids=["zeros", "nan", "small-blocks", "auto-blocks", "uninit"], +) +def test_lazy_special_chunks(arr): + # Run-length chunks live in the offset itself, with no bytes in the file + p = blosc2.open(_put("special.b2nd", arr), lazy=True) + assert p[:].shape == arr.shape + if arr.dtype.kind == "f": + assert np.allclose(p[:], arr[:], equal_nan=True) + + +@pytest.mark.parametrize("clevel", [1, 5, 8, 9]) +def test_lazy_any_clevel(clevel): + # The frame header's flags are a msgpack *string* of raw bytes, and clevel + # rides in the high nibble of one of them: from 8 up it is not valid UTF-8 + a = blosc2.arange(0, 10000, dtype="i4", chunks=(1000,), cparams={"clevel": clevel}) + p = blosc2.open(_put(f"clevel{clevel}.b2nd", a), lazy=True) + assert np.array_equal(p[:], a[:]) + + +def test_lazy_structured_dtype(): + data = np.zeros(1000, dtype=[("a", " +# All rights reserved. +# +# This source code is licensed under a BSD-style license (found in the +# LICENSE file in the root directory of this source tree) +####################################################################### + +"""fsspec reads against a real S3 endpoint, served locally by moto. + +Everything else about the fsspec support is tested over ``memory://``, which is +protocol-generic and needs no dependencies. Two things it structurally cannot +cover, both of which have already hidden a bug: + +- it is not an *async* backend, so ``aget_chunk`` always took its blocking + fallback there, while against s3fs it raised "got Future attached to a + different loop" on every chunk; +- it is poorer in metadata than any real store (no mtime), which let a + size-only cache stamp serve a stale chunk cache. + +These run offline -- moto is a local server, no credentials, no network -- so +they are not marked ``network``. +""" + +import asyncio + +import numpy as np +import pytest + +import blosc2 + +pytest.importorskip("s3fs") +pytest.importorskip("moto") +fsspec = pytest.importorskip("fsspec") + +BUCKET = "blosc2-test" + + +@pytest.fixture(scope="module") +def s3_endpoint(): + """A local S3 server, and fsspec configured to reach it.""" + import fsspec.config + from moto.server import ThreadedMotoServer + + server = ThreadedMotoServer(ip_address="127.0.0.1", port=0, verbose=False) + server.start() + host, port = server.get_host_and_port() + endpoint = f"http://{host}:{port}" + + # blosc2.open() has no storage_options passthrough, so the endpoint and the + # dummy credentials go through fsspec's own per-protocol defaults + previous = fsspec.config.conf.get("s3") + fsspec.config.conf["s3"] = { + "endpoint_url": endpoint, + "key": "testing", + "secret": "testing", + # Not us-east-1: creating a bucket there must carry no location + # constraint, and s3fs sends one whenever it knows the region + "client_kwargs": {"region_name": "eu-west-1"}, + } + fsspec.filesystem("s3", **fsspec.config.conf["s3"]).mkdir(BUCKET) + yield endpoint + + fsspec.config.conf.pop("s3", None) + if previous is not None: + fsspec.config.conf["s3"] = previous + server.stop() + + +@pytest.fixture(scope="module") +def stored(s3_endpoint): + """A 10-chunk array in the bucket, plus the array it was made from.""" + a = blosc2.arange(0, 1000, dtype=np.int32, chunks=(100,)) + urlpath = f"s3://{BUCKET}/ds.b2nd" + a.save(urlpath) + return urlpath, a + + +def test_save_and_open_whole(stored): + urlpath, a = stored + assert np.array_equal(blosc2.open(urlpath)[:], a[:]) + + +def test_cache_storage(stored, tmp_path): + urlpath, a = stored + b = blosc2.open(urlpath, cache_storage=tmp_path, mmap_mode="r") + assert np.array_equal(b[:], a[:]) + + +def test_lazy_range_reads(stored): + urlpath, a = stored + p = blosc2.open(urlpath, lazy=True) + assert np.array_equal(p[150:250], a[150:250]) + assert np.array_equal(p[:], a[:]) + + +@pytest.mark.parametrize("max_concurrency", [1, 8]) +def test_lazy_concurrency(stored, max_concurrency): + urlpath, a = stored + p = blosc2.open(urlpath, lazy=True, max_concurrency=max_concurrency) + assert np.array_equal(p[:], a[:]) + + +@pytest.mark.parametrize("max_concurrency", [1, 8]) +def test_afetch_on_an_async_backend(stored, max_concurrency): + # The regression this file exists for: s3fs runs its coroutines on a private + # event loop, so awaiting one from the caller's loop fails outright + urlpath, a = stored + p = blosc2.open(urlpath, lazy=True) + cache = asyncio.run(p.afetch(slice(150, 250), max_concurrency=max_concurrency)) + assert np.array_equal(cache[150:250], a[150:250]) + + +def test_lazy_expression(stored): + urlpath, a = stored + p = blosc2.open(urlpath, lazy=True) + assert np.array_equal((p * 2)[150:250], a[150:250] * 2) diff --git a/tests/test_schunk.py b/tests/test_schunk.py index 2d2cbe4e4..aed61f499 100644 --- a/tests/test_schunk.py +++ b/tests/test_schunk.py @@ -347,3 +347,34 @@ def test_schunk_update_special(): with pytest.raises(IndexError): schunk2.update_special(5, blosc2.SpecialValue.UNINIT) + + +def test_get_lazychunk_sections(tmp_path): + """A lazy chunk keeps its block offsets and trailer; other chunks stay capped.""" + urlpath = tmp_path / "lazy.b2nd" + a = blosc2.arange( + 0, 100_000, dtype=np.int32, chunks=(50_000,), blocks=(5_000,), urlpath=urlpath, mode="w" + ) + lazychunk = a.schunk.get_lazychunk(0) + nbytes, cbytes, blocksize = blosc2.get_cbuffer_sizes(lazychunk) + nblocks = nbytes // blocksize + assert nblocks == 10 + # Marked lazy, and long enough for header + bstarts + trailer, rather than + # truncated to the header alone + assert lazychunk[31] & 0x08 + assert len(lazychunk) == 32 + 4 * nblocks + 4 + 8 + 4 * nblocks + + # The sections say where every block of the chunk lives, and account for it + bstarts = np.frombuffer(lazychunk[32 : 32 + 4 * nblocks], dtype="= 32 + 4 * nblocks).all() + assert 32 + 4 * nblocks + block_csizes.sum() == cbytes + + # A chunk that is already in memory is not lazy: only its header comes back, + # since copying the whole thing is what asking for a lazy chunk avoids + b = blosc2.arange(0, 100_000, dtype=np.int32, chunks=(50_000,), blocks=(5_000,)) + assert len(b.schunk.get_lazychunk(0)) == blosc2.MAX_OVERHEAD + + # ... but a special chunk still carries the value it repeats + c = blosc2.full((100,), fill_value=np.int32(7)) + assert next(c.schunk.iterchunks_info()).repeated_value == np.int32(7).tobytes()