feat: Handle all forms of range requests

obstore/python/obstore/_get.pyi

@@ -328,19 +328,33 @@ def get_range(
 ) -> Bytes:
     """Return the bytes that are stored at the specified location in the given byte range.
 
-    If the given range is zero-length or starts after the end of the object, an error
-    will be returned. Additionally, if the range ends after the end of the object, the
-    entire remainder of the object will be returned. Otherwise, the exact requested
-    range will be returned.
+    The requested range may be bounded, open-ended, or relative to the end of the
+    object:
+
+    - `start` plus either `end` or `length` requests a specific range of bytes.
+        If the given range is zero-length or starts after the end of the object, an
+        error will be returned. Additionally, if the range ends after the end of the
+        object, the entire remainder of the object will be returned. Otherwise, the
+        exact requested range will be returned.
+    - A non-negative `start` on its own requests all bytes from `start` onwards.
+        This is equivalent to `bytes={start}-` as an HTTP header.
+    - A negative `start` on its own requests the last `abs(start)` bytes. Note that
+        here, `abs(start)` is _the size of the request_, not a byte offset. This is
+        equivalent to `bytes=-{abs(start)}` as an HTTP header.
 
     Args:
         store: The ObjectStore instance to use.
         path: The path within ObjectStore to retrieve.
 
     Keyword Args:
-        start: The start of the byte range.
-        end: The end of the byte range (exclusive). Either `end` or `length` must be non-None.
-        length: The number of bytes of the byte range. Either `end` or `length` must be non-None.
+        start: The start of the byte range. If negative, the last `abs(start)` bytes
+            of the object are requested, and `end` and `length` must both be None.
+        end: The end of the byte range (exclusive). Mutually exclusive with `length`.
+            Defaults to None, in which case the range continues to the end of the
+            object.
+        length: The number of bytes of the byte range. Mutually exclusive with `end`.
+            Defaults to None, in which case the range continues to the end of the
+            object.
 
     Returns:
         A `Bytes` object implementing the Python buffer protocol, allowing
@@ -366,27 +380,53 @@ def get_ranges(
     path: str,
     *,
     starts: Sequence[int],
-    ends: Sequence[int] | None = None,
-    lengths: Sequence[int] | None = None,
-    coalesce: int = 1024 * 1024,
+    ends: Sequence[int | None] | None = None,
+    lengths: Sequence[int | None] | None = None,
+    coalesce: int | None = None,
 ) -> list[Bytes]:
     """Return the bytes stored at the specified location in the given byte ranges.
 
+    Each range is described by one element of `starts` and the element at the same
+    position in `ends` or `lengths`, and follows the same semantics as
+    [get_range][obstore.get_range]. In particular, a negative element of `starts`
+    requests the last `abs(start)` bytes, and a range whose end is left unspecified
+    continues to the end of the object. Because ranges are matched up by position,
+    every sequence given must have the same length as `starts`; a mismatch raises
+    `ValueError`.
+
+    `ends` and `lengths` may both be given at once, so long as at most one of the two
+    is non-None for any individual range — for example `ends=[10, None],
+    lengths=[None, 5]` bounds the first range by its end offset and the second by its
+    length. Omitting both reads every range to the end of the object.
+
     To improve performance this will:
 
     - Transparently combine ranges less than `coalesce` bytes apart into a single
       underlying request (defaults to 1MB)
     - Make multiple `fetch` requests in parallel (up to maximum of 10)
 
+    !!! note
+
+        Combining ranges requires knowing where each one ends, so if any requested
+        range is open-ended or relative to the end of the object, one additional
+        `head` request is made to resolve the size of the object. Requests made up
+        entirely of bounded ranges do not incur this cost.
+
     Args:
         store: The ObjectStore instance to use.
         path: The path within ObjectStore to retrieve.
 
     Other Args:
-        starts: A sequence of `int` where each offset starts.
-        ends: A sequence of `int` where each offset ends (exclusive). Either `ends` or `lengths` must be non-None.
-        lengths: A sequence of `int` with the number of bytes of each byte range. Either `ends` or `lengths` must be non-None.
-        coalesce: Maximum distance in bytes between ranges that will be coalesced into a single request. Defaults to 1MiB. Set to `0` to disable coalescing.
+        starts: A sequence of `int` where each offset starts. A negative value
+            requests the last `abs(start)` bytes for that range, in which case the
+            corresponding elements of `ends` and `lengths` must be None.
+        ends: A sequence of `int` where each offset ends (exclusive). An element may
+            be None to continue that range to the end of the object.
+        lengths: A sequence of `int` with the number of bytes of each byte range. An
+            element may be None to continue that range to the end of the object.
+        coalesce: Maximum distance in bytes between ranges that will be coalesced
+            into a single request. Defaults to None, in which case obstore applies
+            its own default of 1MiB. Set to `0` to disable coalescing.
 
     Returns:
         A sequence of `Bytes`, one for each range. This `Bytes` object implements the
@@ -400,9 +440,9 @@ async def get_ranges_async(
     path: str,
     *,
     starts: Sequence[int],
-    ends: Sequence[int] | None = None,
-    lengths: Sequence[int] | None = None,
-    coalesce: int = 1024 * 1024,
+    ends: Sequence[int | None] | None = None,
+    lengths: Sequence[int | None] | None = None,
+    coalesce: int | None = None,
 ) -> list[Bytes]:
     """Call `get_ranges` asynchronously.
 

obstore/python/obstore/fsspec.py

@@ -35,9 +35,10 @@
 import asyncio
 import warnings
 from collections import defaultdict
+from collections.abc import Iterable
 from functools import cached_property, lru_cache
 from pathlib import Path
-from typing import TYPE_CHECKING, Literal, overload
+from typing import TYPE_CHECKING, Literal, cast, overload
 from urllib.parse import urlparse
 
 import fsspec.asyn
@@ -49,7 +50,7 @@
 
 if TYPE_CHECKING:
     import sys
-    from collections.abc import Coroutine, Iterable
+    from collections.abc import Coroutine, Sequence
     from datetime import datetime
     from typing import Any
 
@@ -113,6 +114,35 @@
 """A type hint for all supported protocols."""
 
 
+def _needs_object_size(start: int | None, end: int | None) -> bool:
+    """Whether a range must be resolved against the object size first.
+
+    A negative `start` on its own is a suffix request, which obstore supports
+    directly. Anything else counting back from the end of the object is not.
+    """
+    return end is not None and (end < 0 or (start is not None and start < 0))
+
+
+def _apply_object_size(
+    start: int | None,
+    end: int | None,
+    size: int | None,
+) -> tuple[int, int | None]:
+    """Rewrite the bounds of a range that count back from the end of an object.
+
+    A `size` of None leaves the range alone, for objects whose size was never
+    needed.
+    """
+    if start is None:
+        start = 0
+    if size is None:
+        return start, end
+    return (
+        max(0, size + start) if start < 0 else start,
+        max(0, size + end) if end is not None and end < 0 else end,
+    )
+
+
 class FsspecStore(fsspec.asyn.AsyncFileSystem):
     """An fsspec implementation based on a obstore Store.
 
@@ -389,12 +419,19 @@ async def _cat_file(
             resp = await store.get_async(path)
             return (await resp.bytes_async()).to_bytes()
 
-        if start is None or end is None:
-            raise NotImplementedError(
-                "cat_file not implemented for start=None xor end=None",
-            )
-
-        range_bytes = await store.get_range_async(path, start=start, end=end)
+        if _needs_object_size(start, end):
+            # See `_resolve_ranges` for why this needs the object size.
+            size = (await store.head_async(path))["size"]
+            if start is not None and start < 0:
+                start = max(0, size + start)
+            if end is not None and end < 0:
+                end = max(0, size + end)
+
+        range_bytes = await store.get_range_async(
+            path,
+            start=0 if start is None else start,
+            end=end,
+        )
         return range_bytes.to_bytes()
 
     async def _cat(  # type: ignore (fsspec has bad typing)
@@ -426,23 +463,30 @@ async def _cat(  # type: ignore (fsspec has bad typing)
     async def _cat_ranges(  # noqa: PLR0913, PLR0917 # type: ignore (fsspec has bad typing)
         self,
         paths: list[str],
-        starts: list[int] | int,
-        ends: list[int] | int,
-        max_gap=None,  # noqa: ANN001, ARG002
-        batch_size=None,  # noqa: ANN001, ARG002
-        on_error="return",  # noqa: ANN001, ARG002
+        starts: Sequence[int | None] | int | None,
+        ends: Sequence[int | None] | int | None,
+        max_gap: int | None = None,
+        batch_size: int | None = None,  # noqa: ARG002
+        on_error: str = "return",  # noqa: ARG002
         **_kwargs: Any,
     ) -> list[bytes]:
-        if isinstance(starts, int):
+        # A non-iterable start or end applies to every path, per fsspec's
+        # AsyncFileSystem._cat_ranges.
+        if not isinstance(starts, Iterable):
             starts = [starts] * len(paths)
-        if isinstance(ends, int):
+        if not isinstance(ends, Iterable):
             ends = [ends] * len(paths)
         if not len(paths) == len(starts) == len(ends):
             raise ValueError
 
-        per_file_requests: dict[str, list[tuple[int, int, int]]] = defaultdict(list)
-        # When upgrading to Python 3.10, use strict=True
-        for idx, (path, start, end) in enumerate(zip(paths, starts, ends)):
+        resolved = await self._resolve_ranges(paths, starts, ends)
+
+        per_file_requests: dict[str, list[tuple[int, int | None, int]]] = defaultdict(
+            list,
+        )
+        for idx, (path, (start, end)) in enumerate(
+            zip(paths, resolved, strict=True),
+        ):
             per_file_requests[path].append((start, end, idx))
 
         futs: list[Coroutine[Any, Any, list[Bytes]]] = []
@@ -451,23 +495,65 @@ async def _cat_ranges(  # noqa: PLR0913, PLR0917 # type: ignore (fsspec has bad
             store = self._construct_store(bucket)
 
             offsets = [r[0] for r in ranges]
-            ends = [r[1] for r in ranges]
-            fut = store.get_ranges_async(path_no_bucket, starts=offsets, ends=ends)
+            file_ends = [r[1] for r in ranges]
+            # fsspec's `max_gap` is obstore's `coalesce`: the largest gap between
+            # two ranges that may still be served by a single request.
+            fut = store.get_ranges_async(
+                path_no_bucket,
+                starts=offsets,
+                ends=file_ends,
+                coalesce=max_gap,
+            )
             futs.append(fut)
 
         result = await asyncio.gather(*futs)
 
         output_buffers: list[bytes] = [b""] * len(paths)
-        # When upgrading to Python 3.10, use strict=True
-        for per_file_request, buffers in zip(per_file_requests.items(), result):
+        for per_file_request, buffers in zip(
+            per_file_requests.items(),
+            result,
+            strict=True,
+        ):
             path, ranges = per_file_request
-            # When upgrading to Python 3.10, use strict=True
-            for buffer, ranges_ in zip(buffers, ranges):
+            for buffer, ranges_ in zip(buffers, ranges, strict=True):
                 initial_index = ranges_[2]
                 output_buffers[initial_index] = buffer.to_bytes()
 
         return output_buffers
 
+    async def _resolve_ranges(
+        self,
+        paths: list[str],
+        starts: Sequence[int | None],
+        ends: Sequence[int | None],
+    ) -> list[tuple[int, int | None]]:
+        """Rewrite fsspec's range vocabulary into the subset obstore accepts.
+
+        `start` is normalized from None to 0. Ranges counting back from the end of
+        the object are resolved against its size, at the cost of one `head` request
+        per object; a lone negative `start` is left alone, since obstore expresses
+        that directly as a suffix request.
+        """
+        sized_paths = sorted(
+            {
+                path
+                for path, start, end in zip(paths, starts, ends, strict=True)
+                if _needs_object_size(start, end)
+            },
+        )
+        sizes = dict(
+            zip(
+                sized_paths,
+                # fsspec types `_sizes` as `list[None]`; the values are really sizes.
+                cast("list[int]", await self._sizes(sized_paths)),
+                strict=True,
+            ),
+        )
+        return [
+            _apply_object_size(start, end, sizes.get(path))
+            for path, start, end in zip(paths, starts, ends, strict=True)
+        ]
+
     async def _put_file(
         self,
         lpath: str,

obstore/python/obstore/store.py

@@ -244,9 +244,9 @@ def get_ranges(
         path: str,
         *,
         starts: Sequence[int],
-        ends: Sequence[int] | None = None,
-        lengths: Sequence[int] | None = None,
-        coalesce: int = 1024 * 1024,
+        ends: Sequence[int | None] | None = None,
+        lengths: Sequence[int | None] | None = None,
+        coalesce: int | None = None,
     ) -> list[Bytes]:
         """Return the bytes stored at the specified location in the given byte ranges.
 
@@ -266,9 +266,9 @@ async def get_ranges_async(
         path: str,
         *,
         starts: Sequence[int],
-        ends: Sequence[int] | None = None,
-        lengths: Sequence[int] | None = None,
-        coalesce: int = 1024 * 1024,
+        ends: Sequence[int | None] | None = None,
+        lengths: Sequence[int | None] | None = None,
+        coalesce: int | None = None,
     ) -> list[Bytes]:
         """Call `get_ranges` asynchronously.