Support fsspec

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

bench/ndarray/fsspec-concurrency.py

@@ -0,0 +1,171 @@
+#!/usr/bin/env python
+
+#######################################################################
+# Copyright (c) 2019-present, Blosc Development Team <blosc@blosc.org>
+# 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://<account>.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()

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
+       <https://filesystem-spec.readthedocs.io>`_ 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
 +++++++++++
 

doc/reference/classes.rst

@@ -139,6 +139,7 @@ container APIs above.
     proxy
     proxysource
     proxyndsource
+    fsspecndsource
     simpleproxy
     embed_store
     dict_store

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

examples/ndarray/concurrent-fsspec.py

@@ -0,0 +1,92 @@
+#######################################################################
+# Copyright (c) 2019-present, Blosc Development Team <blosc@blosc.org>
+# 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])