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

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])

examples/ndarray/rw-fsspec.py

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