feat(storage): add FsspecStore.get_ranges and coalesce_options kwarg

Author: d-v-b

src/zarr/storage/_fsspec.py

@@ -3,6 +3,7 @@
 import json
 import warnings
 from contextlib import suppress
+from functools import partial
 from typing import TYPE_CHECKING, Any
 
 from packaging.version import parse as parse_version
@@ -14,18 +15,24 @@
     Store,
     SuffixByteRequest,
 )
+from zarr.core._coalesce import (
+    DEFAULT_COALESCE_OPTIONS,
+    CoalesceOptions,
+    coalesced_get,
+)
 from zarr.core.buffer import Buffer
 from zarr.errors import ZarrUserWarning
 from zarr.storage._utils import _join_paths, normalize_path
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator, Iterable
+    from collections.abc import AsyncIterator, Iterable, Sequence
 
     from fsspec import AbstractFileSystem
     from fsspec.asyn import AsyncFileSystem
     from fsspec.mapping import FSMap
 
     from zarr.core.buffer import BufferPrototype
+    from zarr.storage._protocols import SupportsGetRanges
 
 
 ALLOWED_EXCEPTIONS: tuple[type[Exception], ...] = (
@@ -124,11 +131,14 @@ def __init__(
         read_only: bool = False,
         path: str = "/",
         allowed_exceptions: tuple[type[Exception], ...] = ALLOWED_EXCEPTIONS,
+        *,
+        coalesce_options: CoalesceOptions = DEFAULT_COALESCE_OPTIONS,
     ) -> None:
         super().__init__(read_only=read_only)
         self.fs = fs
         self.path = normalize_path(path)
         self.allowed_exceptions = allowed_exceptions
+        self.coalesce_options = coalesce_options
 
         if not self.fs.async_impl:
             raise TypeError("Filesystem needs to support async operations.")
@@ -315,6 +325,22 @@ async def get(
         else:
             return value
 
+    async def get_ranges(
+        self,
+        key: str,
+        byte_ranges: Iterable[ByteRequest | None],
+        *,
+        prototype: BufferPrototype,
+    ) -> AsyncIterator[Sequence[tuple[int, Buffer | None]]]:
+        """Read many byte ranges from ``key``, coalescing nearby ranges and fetching concurrently.
+
+        See :class:`zarr.storage._protocols.SupportsGetRanges` for the contract and
+        :func:`zarr.core._coalesce.coalesced_get` for the full semantics.
+        """
+        fetch = partial(self.get, key, prototype)
+        async for group in coalesced_get(fetch, byte_ranges, options=self.coalesce_options):
+            yield group
+
     async def set(
         self,
         key: str,
@@ -440,3 +466,8 @@ async def getsize(self, key: str) -> int:
         else:
             # fsspec doesn't have typing. We'll need to assume or verify this is true
             return int(size)
+
+
+# Module-level type assertion: FsspecStore structurally satisfies SupportsGetRanges.
+# This line is a no-op at runtime but causes mypy/pyright to complain if the shape drifts.
+_: type[SupportsGetRanges] = FsspecStore

tests/test_store/test_fsspec_get_ranges.py

@@ -98,13 +98,13 @@ async def test_coalesce_options_wired_through() -> None:
 
 async def test_get_ranges_mixed_range_types(memory_store: FsspecStore) -> None:
     """Covers RangeByteRequest, OffsetByteRequest, SuffixByteRequest, and None in one call."""
-    from zarr.abc.store import OffsetByteRequest, SuffixByteRequest
+    from zarr.abc.store import ByteRequest, OffsetByteRequest, SuffixByteRequest
 
     blob = bytes(i % 256 for i in range(512))
     await _write(memory_store, "mixed", blob)
     proto = default_buffer_prototype()
 
-    ranges = [
+    ranges: list[ByteRequest | None] = [
         RangeByteRequest(0, 10),
         OffsetByteRequest(500),
         SuffixByteRequest(12),