From 77614389e851b33bcdce5f5df04a266e18ccab1f Mon Sep 17 00:00:00 2001 From: TomNicholas Date: Wed, 1 Jul 2026 17:22:55 -0400 Subject: [PATCH] Add coalescing ManifestStore.get_many Implement Store.get_many on ManifestStore: resolve the requested chunk keys through the manifests to (source file, byte range), group the requests by source file, coalesce each group into runs, and serve each run with a single ranged read that is sliced back into per-chunk buffers. Keys that are not plain manifest-backed chunks (metadata, inlined, or missing chunks) are served individually via `get`. Coalescing uses two knobs (the object_store / async-tiff model): a run merges references whose gap is <= `coalesce_max_gap_bytes` as long as the resulting read stays <= `coalesce_max_bytes`. The gap defaults to 0 - merge only adjacent references, a pure win with no wasted bytes - because benchmarking a 2D map-tile query against a remote Met Office file showed that bridging larger gaps pulls in the chunks that sit between rows (a 2D box is contiguous along one axis but strided along the other), reading ~3x the needed bytes and running slower than no coalescing at all. Merging only adjacent references was ~2.8x faster than the per-chunk baseline with zero over-read. `max_bytes` (default 8 MiB) bounds the size of any single read. Depends on the cross-key `Store.get_many` API in zarr-python (zarr-developers/zarr-python#4112, #4113); until that is released the method is a dormant override. See zarr-developers/VirtualiZarr#947. --- virtualizarr/manifests/store.py | 263 ++++++++++++++-- .../test_manifests/test_store_get_many.py | 292 ++++++++++++++++++ 2 files changed, 534 insertions(+), 21 deletions(-) create mode 100644 virtualizarr/tests/test_manifests/test_store_get_many.py diff --git a/virtualizarr/manifests/store.py b/virtualizarr/manifests/store.py index 6933bd90d..1076c7716 100644 --- a/virtualizarr/manifests/store.py +++ b/virtualizarr/manifests/store.py @@ -1,9 +1,10 @@ from __future__ import annotations +import asyncio import warnings -from collections.abc import AsyncGenerator, Iterable +from collections.abc import AsyncGenerator, Iterable, Sequence from dataclasses import dataclass -from typing import TYPE_CHECKING, Literal, TypeAlias +from typing import TYPE_CHECKING, Literal, NamedTuple, TypeAlias from urllib.parse import urlparse from obspec_utils.registry import ObjectStoreRegistry @@ -44,6 +45,76 @@ class StoreRequest: """The key within the store to request.""" +@dataclass +class _ChunkRef: + """A single chunk request resolved to an absolute byte range in a source file.""" + + store: ObjectStore + path: str + start: int + end: int + + +class _Member(NamedTuple): + """One chunk request within a file group.""" + + request_index: int + """Position of this request in the original ``get_many`` ``requests``.""" + start: int + """Start of the chunk's byte range, absolute within the source file.""" + end: int + """End (exclusive) of the chunk's byte range, absolute within the source file.""" + + +@dataclass +class _FileGroup: + """A set of chunk requests that all read from the same source file.""" + + store: ObjectStore + path: str + members: list[_Member] + + +def _coalesce_members( + members: list[_Member], *, max_gap: int, max_bytes: int +) -> list[list[_Member]]: + """Group members (all in the same file) into runs, each served by one read. + + Two members join the same run when the gap between them is at most + ``max_gap`` *and* the resulting run span stays at most ``max_bytes``. + + The two knobs play distinct roles: + + - ``max_gap`` decides *whether* to bridge a gap between references, i.e. how + much unwanted data coalescing is allowed to pull in. This is what governs + strided access (a large gap bridges the stride of a 2D grid and over-reads; + ``0`` merges only adjacent references and never over-reads). + - ``max_bytes`` caps the *size* of any single read, so a long run of + (near-)adjacent members is split into several bounded reads that the caller + fetches concurrently. It bounds per-read memory and lets a large contiguous + span be fetched over several parallel connections rather than one. It does + *not* affect over-read. + + Runs are returned sorted by start offset and cover every member exactly + once. A member larger than ``max_bytes`` still forms its own (single) run. + """ + ordered = sorted(members, key=lambda member: member.start) + runs: list[list[_Member]] = [] + run_start = run_end = 0 + for member in ordered: + if ( + runs + and member.start - run_end <= max_gap + and max(run_end, member.end) - run_start <= max_bytes + ): + runs[-1].append(member) + run_end = max(run_end, member.end) + else: + runs.append([member]) + run_start, run_end = member.start, member.end + return runs + + def get_store_prefix(url: str) -> str: """ Get a logical prefix to use for a url in an ObjectStoreRegistry @@ -112,7 +183,12 @@ def __eq__(self, value: object): NotImplementedError def __init__( - self, group: ManifestGroup, *, registry: ObjectStoreRegistry | None = None + self, + group: ManifestGroup, + *, + registry: ObjectStoreRegistry | None = None, + coalesce_max_gap_bytes: int = 0, + coalesce_max_bytes: int = 8 * 1024 * 1024, ) -> None: """Instantiate a new ManifestStore. @@ -123,6 +199,27 @@ def __init__( registry A registry mapping the URL scheme and netloc to [ObjectStore][obstore.store.ObjectStore] instances, allowing [ManifestStores][virtualizarr.manifests.ManifestStore] to read from different [ObjectStore][obstore.store.ObjectStore] instances. + coalesce_max_gap_bytes + When multiple chunks requested together (via ``get_many``) refer to the + same source file, virtual references separated by at most this many bytes + are coalesced into a single, larger read. Defaults to ``0``, which merges + only references that are exactly adjacent in the file - a pure win, since + no unwanted bytes are read. A larger value additionally bridges gaps up to + that size, trading some wasted bytes for fewer requests; this can backfire + for strided access patterns (e.g. a 2D spatial box, whose chunks are + contiguous along one axis but far apart along another), where bridging the + stride pulls in the intervening chunks. See + [ManifestStore.get_many][virtualizarr.manifests.ManifestStore.get_many]. + coalesce_max_bytes + Upper bound on the size of a single coalesced read. A run of adjacent (or, + with a non-zero gap, near-adjacent) references longer than this is split + into several reads that are fetched concurrently, bounding per-read memory + and letting a large contiguous span be pulled over several parallel + connections instead of one. This is the equivalent of icechunk's + ``ideal_concurrent_request_size``, provided here because obstore does not + split a single ranged read on its own. It does not affect over-read (that + is ``coalesce_max_gap_bytes``), and is inert unless a query coalesces into a + run larger than this. Defaults to 8 MiB. """ if not isinstance(group, ManifestGroup): @@ -131,6 +228,8 @@ def __init__( super().__init__(read_only=True) self._registry = ObjectStoreRegistry() if registry is None else registry self._group = group + self._coalesce_max_gap_bytes = coalesce_max_gap_bytes + self._coalesce_max_bytes = coalesce_max_bytes def __str__(self) -> str: return f"ManifestStore(group={self._group}, registry={self._registry})" @@ -182,16 +281,30 @@ async def get( entry = manifest.get_entry(chunk_indexes) if entry is None: return None - path = entry["path"] - offset = entry["offset"] - length = entry["length"] - # Get the configured object store instance that matches the path - store, path_after_prefix = self._registry.resolve(path) + store, path_in_store = self._resolve_store_and_path(entry["path"]) + # Transform the input byte range to account for the chunk location in the file + byte_range = _transform_byte_range( + byte_range, + chunk_start=entry["offset"], + chunk_end_exclusive=entry["offset"] + entry["length"], + ) + + # Actually get the bytes + bytes = await store.get_range_async( + path_in_store, + start=byte_range.start, + end=byte_range.end, + ) + return prototype.buffer.from_bytes(bytes) # type: ignore[arg-type] + + def _resolve_store_and_path(self, path: str) -> tuple[ObjectStore, str]: + """Resolve a manifest entry ``path`` to its ObjectStore and the path + within that store (with any store prefix stripped).""" + store, _ = self._registry.resolve(path) if not store: raise ValueError( f"Could not find a store to use for {path} in the store registry" ) - path_in_store = urlparse(path).path if hasattr(store, "prefix") and store.prefix: prefix = str(store.prefix).lstrip("/") @@ -199,20 +312,39 @@ async def get( prefix = urlparse(store.url).path.lstrip("/") else: prefix = "" - path_in_store = path_in_store.lstrip("/").removeprefix(prefix).lstrip("/") - # Transform the input byte range to account for the chunk location in the file - chunk_end_exclusive = offset + length - byte_range = _transform_byte_range( - byte_range, chunk_start=offset, chunk_end_exclusive=chunk_end_exclusive - ) + return store, path_in_store.lstrip("/").removeprefix(prefix).lstrip("/") - # Actually get the bytes - bytes = await store.get_range_async( - path_in_store, - start=byte_range.start, - end=byte_range.end, + def _resolve_chunk_ref( + self, key: str, byte_range: ByteRequest | None + ) -> _ChunkRef | None: + """Resolve a chunk key to an absolute byte range within its source file. + + Returns ``None`` when the key is not a plain, manifest-backed chunk that + can participate in coalescing - i.e. metadata documents, inlined chunks, + missing chunks, or group keys. Those are handled individually by ``get``. + """ + node, suffix = _get_deepest_group_or_array(self._group, key) + if suffix.endswith( + ("zarr.json", ".zattrs", ".zgroup", ".zarray", ".zmetadata") + ) or isinstance(node, ManifestGroup): + return None + manifest = node.manifest + separator: Literal[".", "/"] = getattr( + node.metadata.chunk_key_encoding, "separator", "." ) - return prototype.buffer.from_bytes(bytes) # type: ignore[arg-type] + chunk_indexes = parse_manifest_index(key, separator, expand_pattern=True) + if chunk_indexes in manifest._inlined: + return None + entry = manifest.get_entry(chunk_indexes) + if entry is None: + return None + store, path_in_store = self._resolve_store_and_path(entry["path"]) + rng = _transform_byte_range( + byte_range, + chunk_start=entry["offset"], + chunk_end_exclusive=entry["offset"] + entry["length"], + ) + return _ChunkRef(store=store, path=path_in_store, start=rng.start, end=rng.end) async def get_partial_values( self, @@ -223,6 +355,95 @@ async def get_partial_values( # TODO: Implement using private functions from the upstream Zarr obstore integration raise NotImplementedError + async def get_many( + self, + requests: Sequence[tuple[str, ByteRequest | None] | str], + *, + prototype: BufferPrototype, + ) -> AsyncGenerator[Sequence[tuple[int, Buffer | None]], None]: + """Retrieve many chunks at once, coalescing reads from the same file. + + Overrides [Store.get_many][zarr.abc.store.Store.get_many]. Requested + chunks are resolved through the manifests to ``(source file, byte + range)`` and grouped by source file. Within each file, references closer + together than ``coalesce_max_gap_bytes`` are coalesced into runs; each run + is then split into reads of at most ``coalesce_max_bytes`` and served by a + ranged read that is sliced back into per-chunk buffers. The gap controls + how much unwanted data coalescing may pull in - ``0`` (the default) merges + only adjacent references and never over-reads, which matters for strided + access such as a 2D spatial box - while the size cap only bounds how large + any single read may get (splitting big contiguous runs into parallel + reads); it does not affect over-read. + This is the same technique object_store and async-tiff use to read many + tiles efficiently, applied here to virtual chunk references. + + Keys that are not plain manifest-backed chunks (metadata documents, + inlined chunks, or missing chunks) are served individually via ``get``. + Results are yielded as ``(request_index, Buffer | None)`` batches in + completion order, per the ``Store.get_many`` contract. + """ + # Local import so importing this module doesn't require the zarr config. + from zarr.core.config import config + + # Partition requests into coalescable chunk reads (grouped by source + # file) and everything else (served one-by-one via ``get``). + groups: dict[tuple[int, str], _FileGroup] = {} + singletons: list[tuple[int, str, ByteRequest | None]] = [] + for index, request in enumerate(requests): + key, byte_range = (request, None) if isinstance(request, str) else request + ref = self._resolve_chunk_ref(key, byte_range) + if ref is None: + singletons.append((index, key, byte_range)) + continue + group = groups.setdefault( + (id(ref.store), ref.path), _FileGroup(ref.store, ref.path, []) + ) + group.members.append(_Member(index, ref.start, ref.end)) + + semaphore = asyncio.Semaphore(config.get("async.concurrency")) + + async def fetch_run( + store: ObjectStore, path: str, run: list[_Member] + ) -> Sequence[tuple[int, Buffer | None]]: + # One ranged read spanning the whole run, sliced back per chunk. + run_start = run[0].start + run_end = max(member.end for member in run) + async with semaphore: + data = await store.get_range_async(path, start=run_start, end=run_end) + view = memoryview(data) # type: ignore[arg-type] + return [ + ( + member.request_index, + prototype.buffer.from_bytes( + view[member.start - run_start : member.end - run_start] + ), + ) + for member in run + ] + + async def fetch_single( + index: int, key: str, byte_range: ByteRequest | None + ) -> Sequence[tuple[int, Buffer | None]]: + async with semaphore: + buffer = await self.get(key, prototype, byte_range) + return ((index, buffer),) + + tasks = [ + asyncio.ensure_future(fetch_run(group.store, group.path, run)) + for group in groups.values() + for run in _coalesce_members( + group.members, + max_gap=self._coalesce_max_gap_bytes, + max_bytes=self._coalesce_max_bytes, + ) + ] + tasks += [ + asyncio.ensure_future(fetch_single(index, key, byte_range)) + for index, key, byte_range in singletons + ] + for coro in asyncio.as_completed(tasks): + yield await coro + async def exists(self, key: str) -> bool: # docstring inherited raise NotImplementedError diff --git a/virtualizarr/tests/test_manifests/test_store_get_many.py b/virtualizarr/tests/test_manifests/test_store_get_many.py new file mode 100644 index 000000000..38ae9552d --- /dev/null +++ b/virtualizarr/tests/test_manifests/test_store_get_many.py @@ -0,0 +1,292 @@ +from __future__ import annotations + +import numpy as np +import pytest +from obspec_utils.registry import ObjectStoreRegistry +from zarr.core.buffer import default_buffer_prototype + +from virtualizarr.manifests import ( + ChunkManifest, + ManifestArray, + ManifestGroup, + ManifestStore, +) +from virtualizarr.manifests.store import _coalesce_members, _Member +from virtualizarr.manifests.utils import create_v3_array_metadata +from virtualizarr.tests import requires_obstore + +# 16 distinct bytes, enough for four contiguous 4-byte chunks. +FILE_BYTES = bytes(range(1, 17)) +_CODECS = [{"configuration": {"endian": "little"}, "name": "bytes"}] + + +def _build_store( + tmpdir, + *, + files: dict[str, bytes] | None = None, + entries: dict[str, dict] | None = None, + coalesce_max_gap_bytes: int = 1024 * 1024, + coalesce_max_bytes: int = 8 * 1024 * 1024, +) -> ManifestStore: + """Build a ManifestStore over a LocalStore with one array ``foo`` (4x4, 2x2 chunks).""" + import obstore as obs + + store = obs.store.LocalStore() + prefix = "file://" + files = files or {f"{tmpdir}/data.tmp": FILE_BYTES} + for filepath, data in files.items(): + obs.put(store, filepath, data) + + if entries is None: + fp = next(iter(files)) + entries = { + "0.0": {"path": f"{prefix}/{fp}", "offset": 0, "length": 4}, + "0.1": {"path": f"{prefix}/{fp}", "offset": 4, "length": 4}, + "1.0": {"path": f"{prefix}/{fp}", "offset": 8, "length": 4}, + "1.1": {"path": f"{prefix}/{fp}", "offset": 12, "length": 4}, + } + manifest = ChunkManifest(entries=entries) + metadata = create_v3_array_metadata( + shape=(4, 4), + chunk_shape=(2, 2), + data_type=np.dtype("int32"), + codecs=_CODECS, + chunk_key_encoding={"name": "default", "separator": "."}, + fill_value=0, + ) + marr = ManifestArray(metadata=metadata, chunkmanifest=manifest) + group = ManifestGroup(arrays={"foo": marr}) + registry = ObjectStoreRegistry({prefix: store}) + return ManifestStore( + group=group, + registry=registry, + coalesce_max_gap_bytes=coalesce_max_gap_bytes, + coalesce_max_bytes=coalesce_max_bytes, + ) + + +async def _collect(store: ManifestStore, requests) -> dict[int, object]: + """Drain get_many into an index -> buffer mapping, asserting each index appears once.""" + collected: dict[int, object] = {} + async for batch in store.get_many(requests, prototype=default_buffer_prototype()): + for index, buffer in batch: + assert index not in collected, "each request must be reported exactly once" + collected[index] = buffer + return collected + + +class _RecordingStore: + """Proxy that records each ranged read (``get_range_async``) and forwards + everything else. + + Installed by monkeypatching ``ManifestStore._resolve_store_and_path`` so we + can see exactly which byte ranges get_many issues after coalescing. + """ + + def __init__(self, inner, log: list[dict]) -> None: + self._inner = inner + self._log = log + + def __getattr__(self, name): + return getattr(self._inner, name) + + async def get_range_async(self, path, *, start, end): + self._log.append({"path": path, "start": start, "end": end}) + return await self._inner.get_range_async(path, start=start, end=end) + + +@pytest.fixture() +def record_range_calls(monkeypatch): + """Record every get_ranges_async call ManifestStore.get_many issues.""" + log: list[dict] = [] + original = ManifestStore._resolve_store_and_path + proxies: dict[int, _RecordingStore] = {} + + def patched(self, path): + inner, path_in_store = original(self, path) + proxy = proxies.get(id(inner)) + if proxy is None: + proxy = _RecordingStore(inner, log) + proxies[id(inner)] = proxy + return proxy, path_in_store + + monkeypatch.setattr(ManifestStore, "_resolve_store_and_path", patched) + return log + + +@requires_obstore +class TestGetMany: + pytestmark = pytest.mark.asyncio + + async def test_matches_individual_get(self, tmpdir): + store = _build_store(tmpdir) + keys = ["foo/c.0.0", "foo/c.0.1", "foo/c.1.0", "foo/c.1.1"] + collected = await _collect(store, keys) + + assert set(collected) == set(range(len(keys))) + for index, key in enumerate(keys): + expected = await store.get(key, prototype=default_buffer_prototype()) + assert collected[index].to_bytes() == expected.to_bytes() + # sanity: the four chunks are the four consecutive 4-byte slices + assert collected[0].to_bytes() == FILE_BYTES[0:4] + assert collected[3].to_bytes() == FILE_BYTES[12:16] + + async def test_accepts_key_range_tuples(self, tmpdir): + from zarr.abc.store import RangeByteRequest + + store = _build_store(tmpdir) + requests = [ + "foo/c.0.0", + ("foo/c.0.1", None), + ("foo/c.1.0", RangeByteRequest(1, 3)), # middle two bytes of that chunk + ] + collected = await _collect(store, requests) + assert collected[0].to_bytes() == FILE_BYTES[0:4] + assert collected[1].to_bytes() == FILE_BYTES[4:8] + assert collected[2].to_bytes() == FILE_BYTES[8:12][1:3] + + async def test_missing_chunk_is_none(self, tmpdir): + prefix = "file://" + fp = f"{tmpdir}/data.tmp" + # omit chunk "1.1" so it has no manifest entry + entries = { + "0.0": {"path": f"{prefix}/{fp}", "offset": 0, "length": 4}, + "0.1": {"path": f"{prefix}/{fp}", "offset": 4, "length": 4}, + "1.0": {"path": f"{prefix}/{fp}", "offset": 8, "length": 4}, + } + store = _build_store(tmpdir, entries=entries) + collected = await _collect(store, ["foo/c.0.0", "foo/c.1.1"]) + assert collected[0].to_bytes() == FILE_BYTES[0:4] + assert collected[1] is None # missing -> None, not omitted + + async def test_metadata_key_served_individually(self, tmpdir): + store = _build_store(tmpdir) + collected = await _collect(store, ["foo/zarr.json", "foo/c.0.0"]) + assert collected[0] is not None # zarr.json metadata document + assert b'"zarr_format"' in collected[0].to_bytes() + assert collected[1].to_bytes() == FILE_BYTES[0:4] + + async def test_empty_requests(self, tmpdir): + store = _build_store(tmpdir) + collected = await _collect(store, []) + assert collected == {} + + async def test_coalesces_contiguous_into_one_read(self, tmpdir, record_range_calls): + store = _build_store(tmpdir) + keys = ["foo/c.0.0", "foo/c.0.1", "foo/c.1.0", "foo/c.1.1"] + collected = await _collect(store, keys) + + # the four contiguous chunks are served by a single ranged read [0, 16) + assert [(c["start"], c["end"]) for c in record_range_calls] == [(0, 16)] + # and the bytes were sliced back to the right chunks + assert [collected[i].to_bytes() for i in range(4)] == [ + FILE_BYTES[0:4], + FILE_BYTES[4:8], + FILE_BYTES[8:12], + FILE_BYTES[12:16], + ] + + async def test_groups_requests_by_file(self, tmpdir, record_range_calls): + prefix = "file://" + fp_a = f"{tmpdir}/a.tmp" + fp_b = f"{tmpdir}/b.tmp" + entries = { + "0.0": {"path": f"{prefix}/{fp_a}", "offset": 0, "length": 4}, + "0.1": {"path": f"{prefix}/{fp_a}", "offset": 4, "length": 4}, + "1.0": {"path": f"{prefix}/{fp_b}", "offset": 0, "length": 4}, + "1.1": {"path": f"{prefix}/{fp_b}", "offset": 4, "length": 4}, + } + store = _build_store( + tmpdir, files={fp_a: FILE_BYTES, fp_b: FILE_BYTES}, entries=entries + ) + await _collect(store, ["foo/c.0.0", "foo/c.0.1", "foo/c.1.0", "foo/c.1.1"]) + + # one read per distinct source file, each spanning that file's two chunks + assert len(record_range_calls) == 2 + for call in record_range_calls: + assert (call["start"], call["end"]) == (0, 8) + + async def test_gap_larger_than_max_gap_is_not_merged( + self, tmpdir, record_range_calls + ): + prefix = "file://" + fp = f"{tmpdir}/data.tmp" + # two chunks 996 bytes apart in the file + entries = { + "0.0": {"path": f"{prefix}/{fp}", "offset": 0, "length": 4}, + "0.1": {"path": f"{prefix}/{fp}", "offset": 1000, "length": 4}, + } + store = _build_store( + tmpdir, + files={fp: bytes(2048)}, + entries=entries, + coalesce_max_gap_bytes=8, # < 996 gap -> keep separate + ) + await _collect(store, ["foo/c.0.0", "foo/c.0.1"]) + assert sorted((c["start"], c["end"]) for c in record_range_calls) == [ + (0, 4), + (1000, 1004), + ] + + async def test_max_bytes_caps_a_merge(self, tmpdir, record_range_calls): + prefix = "file://" + fp = f"{tmpdir}/data.tmp" + entries = { + "0.0": {"path": f"{prefix}/{fp}", "offset": 0, "length": 4}, + "0.1": {"path": f"{prefix}/{fp}", "offset": 500, "length": 4}, + } + # the 500-byte gap is within max_gap, but merging would make a 504-byte + # read which exceeds max_bytes, so the two chunks stay separate. + store = _build_store( + tmpdir, + files={fp: bytes(2048)}, + entries=entries, + coalesce_max_gap_bytes=1000, + coalesce_max_bytes=100, + ) + await _collect(store, ["foo/c.0.0", "foo/c.0.1"]) + assert sorted((c["start"], c["end"]) for c in record_range_calls) == [ + (0, 4), + (500, 504), + ] + + +@requires_obstore +class TestCoalesceMembers: + def _members(self, *ranges): + return [_Member(i, start, end) for i, (start, end) in enumerate(ranges)] + + def test_merges_contiguous_and_near(self): + # 0-4 and 4-8 touch; 8-12 is 2 bytes after -> all merge with gap>=2 + runs = _coalesce_members( + self._members((0, 4), (6, 10)), max_gap=2, max_bytes=1 << 20 + ) + assert len(runs) == 1 + assert [(m.start, m.end) for m in runs[0]] == [(0, 4), (6, 10)] + + def test_splits_on_large_gap(self): + runs = _coalesce_members( + self._members((0, 4), (100, 104)), max_gap=8, max_bytes=1 << 20 + ) + assert [[(m.start, m.end) for m in run] for run in runs] == [ + [(0, 4)], + [(100, 104)], + ] + + def test_splits_when_span_exceeds_max_bytes(self): + runs = _coalesce_members( + self._members((0, 4), (50, 54)), max_gap=1000, max_bytes=40 + ) + assert len(runs) == 2 + + def test_oversized_single_member_is_its_own_run(self): + runs = _coalesce_members(self._members((0, 1000)), max_gap=8, max_bytes=100) + assert len(runs) == 1 + assert runs[0][0].end == 1000 + + def test_sorts_by_start(self): + runs = _coalesce_members( + self._members((8, 12), (0, 4), (4, 8)), max_gap=0, max_bytes=1 << 20 + ) + # all adjacent -> one run, ordered by start + assert [(m.start, m.end) for m in runs[0]] == [(0, 4), (4, 8), (8, 12)]