From 9bfe855a6cfd6985073fd5e4e5b5be027628eca8 Mon Sep 17 00:00:00 2001 From: Davis Bennett Date: Tue, 14 Jul 2026 13:25:16 +0200 Subject: [PATCH 1/4] fix: byte-order handling for structured dtypes in the bytes codec (#220) * fix: byte-order handling for structured dtypes in the bytes codec The bytes codec neither byte-swapped structured-dtype fields to its configured endian on encode (numpy reports byteorder '|' for void dtypes, so the top-level byteorder comparison never detected a mismatch) nor honored its endian when decoding, silently corrupting any structured data whose field byte order differed from the stored one (e.g. virtual references to external big-endian data). Encode now detects byte-order mismatches by comparing full dtypes via newbyteorder, and decode reinterprets raw bytes in the stored byte order before converting to the data type's declared byte order, so the stored layout (codec state) and the in-memory layout (array data type) are independent. Closes #4141 Assisted-by: ClaudeCode:claude-fable-5 * test: fold structured byte-order cases into existing bytes codec tests Extend test_endian's parametrization with structured dtypes and test_bytes_codec_sync_roundtrip with endian/dtype parametrization plus stored-layout and decoded-dtype assertions, instead of adding parallel test functions for the same properties. Assisted-by: ClaudeCode:claude-fable-5 * refactor: rename stored_dtype to view_dtype in BytesCodec decode The variable is the dtype used to view the raw chunk bytes (byte order from the codec's endian configuration), not a property of the stored data or of the returned buffer, which always carries the array's declared dtype. Assisted-by: ClaudeCode:claude-fable-5 * docs: note that the decode-side byte-order conversion copies the chunk Assisted-by: ClaudeCode:claude-fable-5 --- changes/4141.bugfix.md | 1 + src/zarr/codecs/bytes.py | 33 +++++++++++----- tests/test_codecs/test_bytes.py | 69 ++++++++++++++++++++++++++------- 3 files changed, 80 insertions(+), 23 deletions(-) create mode 100644 changes/4141.bugfix.md diff --git a/changes/4141.bugfix.md b/changes/4141.bugfix.md new file mode 100644 index 0000000000..6a132da3f5 --- /dev/null +++ b/changes/4141.bugfix.md @@ -0,0 +1 @@ +Fix silent byte-order corruption for structured dtypes with the `bytes` codec: multi-byte fields are now byte-swapped to the codec's configured `endian` on write and decoded honoring it on read, so non-native-endian structured data (e.g. big-endian fields, as produced by virtual references to external data) round-trips correctly. diff --git a/src/zarr/codecs/bytes.py b/src/zarr/codecs/bytes.py index 240c077627..fae762fd08 100644 --- a/src/zarr/codecs/bytes.py +++ b/src/zarr/codecs/bytes.py @@ -100,14 +100,28 @@ def _decode_sync( chunk_spec: ArraySpec, ) -> NDBuffer: endian_str = self.endian + dtype = chunk_spec.dtype.to_native_dtype() + # The byte order of the stored data is set by this codec's `endian` + # configuration; the byte order of the decoded array is set by the array's + # data type. The two are independent: the raw bytes are viewed with a dtype + # in the stored byte order, then converted to the declared dtype if needed. if isinstance(chunk_spec.dtype, HasEndianness): - dtype = replace(chunk_spec.dtype, endianness=endian_str).to_native_dtype() # type: ignore[call-arg] + view_dtype = replace(chunk_spec.dtype, endianness=endian_str).to_native_dtype() # type: ignore[call-arg] + elif isinstance(chunk_spec.dtype, Struct) and endian_str is not None: + # Per the struct data type spec, all multi-byte fields are stored in the + # byte order configured on this codec. + view_dtype = dtype.newbyteorder(endian_str) else: - dtype = chunk_spec.dtype.to_native_dtype() + view_dtype = dtype as_array_like = chunk_bytes.as_array_like() chunk_array = chunk_spec.prototype.nd_buffer.from_ndarray_like( - as_array_like.view(dtype=dtype) # type: ignore[attr-defined] + as_array_like.view(dtype=view_dtype) # type: ignore[attr-defined] ) + if view_dtype != dtype: + # This byte-swapping conversion copies the chunk. The dtype inequality + # guard keeps the common case, where the stored and declared byte orders + # already match, on the zero-copy view path above. + chunk_array = chunk_array.astype(dtype) # ensure correct chunk shape if chunk_array.shape != chunk_spec.shape: @@ -129,13 +143,14 @@ def _encode_sync( chunk_spec: ArraySpec, ) -> Buffer | None: assert isinstance(chunk_array, NDBuffer) - if ( - chunk_array.dtype.itemsize > 1 - and self.endian is not None - and self.endian != chunk_array.byteorder - ): + if chunk_array.dtype.itemsize > 1 and self.endian is not None: + # Compare full dtypes rather than the top-level byteorder: numpy reports + # byteorder '|' for structured dtypes even when their fields are + # byte-order-sensitive, so newbyteorder is the only reliable way to + # detect (and normalize) a byte-order mismatch. new_dtype = chunk_array.dtype.newbyteorder(self.endian) - chunk_array = chunk_array.astype(new_dtype) + if new_dtype != chunk_array.dtype: + chunk_array = chunk_array.astype(new_dtype) nd_array = chunk_array.as_ndarray_like() # Flatten the nd-array (only copy if needed) and reinterpret as bytes diff --git a/tests/test_codecs/test_bytes.py b/tests/test_codecs/test_bytes.py index 03dd0b40c6..ead778f526 100644 --- a/tests/test_codecs/test_bytes.py +++ b/tests/test_codecs/test_bytes.py @@ -33,29 +33,50 @@ @pytest.mark.parametrize("store", ["local", "memory"], indirect=["store"]) -@pytest.mark.parametrize("input_dtype", [">u2", "u2", + "f4"), ("mask", ">i4")], + [("flux", "u2", " None: """ The `bytes` codec stores multi-byte data in the byte order configured on the codec, regardless of the input array's byte order, and reads it back to the - original values. The input-dtype/store-endian cross-product exercises the - encode-side byteswap (input byte order != store byte order) and the no-op - case alike. Compression is disabled so the stored chunk is the codec's raw - output and its byte layout can be asserted directly. + original values. For structured dtypes this applies to every multi-byte + field, per the `struct` data type spec; the struct cases guard against the + endianness bugs from + https://github.com/zarr-developers/zarr-python/issues/4141, where the + encode path never byte-swapped struct fields (numpy reports byteorder '|' + for void dtypes) and the decode path ignored the codec's endian entirely. + The input-dtype/store-endian cross-product exercises the encode-side + byteswap (input byte order != store byte order) and the no-op case alike. + Compression is disabled so the stored chunk is the codec's raw output and + its byte layout can be asserted directly. """ - data = np.arange(0, 256, dtype=input_dtype).reshape((16, 16)) + dtype = np.dtype(input_dtype) + if dtype.fields is None: + data = np.arange(0, 256, dtype=dtype).reshape((16, 16)) + else: + data = np.zeros((16, 16), dtype=dtype) + data["flux"] = np.arange(0, 256).reshape((16, 16)) + data["mask"] = np.arange(256, 512).reshape((16, 16)) path = "endian" spath = StorePath(store, path) a = await zarr.api.asynchronous.create_array( spath, shape=data.shape, chunks=(16, 16), - dtype="uint16", + dtype=dtype, fill_value=0, compressors=None, serializer=BytesCodec(endian=store_endian), @@ -66,8 +87,7 @@ async def test_endian( # The stored chunk is laid out in the byte order configured on the codec. stored = await store.get(f"{path}/c/0/0", prototype=default_buffer_prototype()) assert stored is not None - expected_dtype = ">u2" if store_endian == "big" else " None: assert isinstance(BytesCodec(), SupportsSyncCodec) -def test_bytes_codec_sync_roundtrip() -> None: - codec = BytesCodec() - arr = np.arange(100, dtype="float64") +@pytest.mark.parametrize("endian", ENDIAN) +@pytest.mark.parametrize( + "native_dtype", + [np.dtype("float64"), np.dtype(">u2"), np.dtype([("a", ">f4"), ("b", " None: + """ + The synchronous encode/decode path round-trips data, and the two byte + orders involved are independent: the codec's `endian` configuration governs + only the stored byte layout (every multi-byte value, including struct + fields, is laid out in the codec's byte order regardless of the input + array's byte order), while the decoded buffer's byte order is governed by + the array's data type regardless of the codec's. The mixed-endian struct + case pins that per-field byte order of the in-memory dtype survives a + roundtrip through a single stored byte order. + """ + if native_dtype.fields is None: + arr = np.arange(100, dtype=native_dtype) + else: + arr = np.array([(1.5, 2), (3.5, 4), (5.5, 6), (7.5, 8)], dtype=native_dtype) zdtype = get_data_type_from_native_dtype(arr.dtype) spec = ArraySpec( shape=arr.shape, @@ -91,11 +129,14 @@ def test_bytes_codec_sync_roundtrip() -> None: ) nd_buf: NDBuffer = default_buffer_prototype().nd_buffer.from_numpy_array(arr) - codec = codec.evolve_from_array_spec(spec) + codec = BytesCodec(endian=endian).evolve_from_array_spec(spec) encoded = codec._encode_sync(nd_buf, spec) assert encoded is not None + assert encoded.to_bytes() == arr.astype(native_dtype.newbyteorder(endian)).tobytes() + decoded = codec._decode_sync(encoded, spec) + assert decoded.dtype == zdtype.to_native_dtype() np.testing.assert_array_equal(arr, decoded.as_numpy_array()) From aa766b41bbd4135e5ae701c615503ef16b64647a Mon Sep 17 00:00:00 2001 From: Davis Vann Bennett Date: Tue, 28 Jul 2026 07:13:12 +0200 Subject: [PATCH 2/4] fix(codec_pipeline): keep FusedCodecPipeline compute off the event-loop thread MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FusedCodecPipeline.read/write ran their synchronous fast path inline on the coroutine servicing the request — i.e. on the global zarr_io event loop thread. Single-chunk batches decoded inline on the loop and multi-chunk batches blocked the loop in pool.map, so every sync-API call from every user thread serialized behind each other's codec compute. The blocked window scales with codec cost, which is why users reported the fused pipeline as "slower for zstd-compressed data" under multi-threaded (dask-style, one chunk per call) access: at 8 reader threads on 4 MiB zstd chunks it was 3.4x slower than BatchedCodecPipeline, and throughput did not scale with threads at all (336 -> 439 ms from 1 to 8 threads, versus 669 -> 121 ms for batched). Offload the synchronous batch to a worker thread with asyncio.to_thread: one hop per batch, not per chunk, preserving the fused pipeline's win over per-chunk async scheduling while keeping the loop free. After the fix the same workload scales 625 -> 109 ms from 1 to 8 threads, beating batched at every thread count; single-threaded performance is unchanged (the hop costs ~75 us per batch). The regression test asserts deterministically (no timing) that codec compute never runs on a thread with a running event loop, covering single- and multi-chunk reads and writes through the sync API. A new benchmark covers the many-threads/one-chunk-per-call access pattern. Assisted-by: ClaudeCode:claude-fable-5 --- changes/247.bugfix.md | 10 +++++++ src/zarr/core/codec_pipeline.py | 16 ++++++++-- tests/benchmarks/test_e2e.py | 48 +++++++++++++++++++++++++++++ tests/test_fused_pipeline.py | 53 +++++++++++++++++++++++++++++++++ 4 files changed, 125 insertions(+), 2 deletions(-) create mode 100644 changes/247.bugfix.md diff --git a/changes/247.bugfix.md b/changes/247.bugfix.md new file mode 100644 index 0000000000..21a4924664 --- /dev/null +++ b/changes/247.bugfix.md @@ -0,0 +1,10 @@ +`FusedCodecPipeline` no longer runs chunk IO and codec compute on the thread +driving zarr's internal event loop. Previously each read/write executed its +synchronous fast path inline on that loop thread, and because every sync-API +call from every user thread is serviced by the same loop, concurrent +operations serialized behind each other's codec work — reported as the fused +pipeline being slower than `BatchedCodecPipeline` for zstd-compressed data +under multi-threaded (e.g. dask) access. The synchronous batch now runs on a +worker thread (one hop per batch, not per chunk), keeping the loop free. +Multi-threaded single-chunk reads of zstd data are ~4.5x faster than before +and now scale with reader threads; single-threaded performance is unchanged. diff --git a/src/zarr/core/codec_pipeline.py b/src/zarr/core/codec_pipeline.py index 2e3f1ed122..4b8831bc7b 100644 --- a/src/zarr/core/codec_pipeline.py +++ b/src/zarr/core/codec_pipeline.py @@ -1182,7 +1182,15 @@ async def read( (isinstance(first_bg, StorePath) and isinstance(first_bg.store, SupportsGetSync)) or (not isinstance(first_bg, StorePath) and isinstance(first_bg, SyncByteGetter)) ): - return self.read_sync(batch, out, drop_axes, max_workers=_resolve_max_workers()) + # One thread hop for the WHOLE batch — not per chunk, so the fused + # design's win over per-chunk async scheduling is preserved. Running + # read_sync inline here would block the event loop for the duration + # of the batch's IO+compute; every sync-API call from every user + # thread shares this one loop, so inline execution serializes + # concurrent callers behind each other's codec compute. + return await asyncio.to_thread( + self.read_sync, batch, out, drop_axes, max_workers=_resolve_max_workers() + ) # Non-sync store (e.g. ZipStore): can't use the sync fast path. But if the # array-bytes codec supports partial decoding (sharding), still route @@ -1235,7 +1243,11 @@ async def write( (isinstance(first_bs, StorePath) and isinstance(first_bs.store, SupportsSetSync)) or (not isinstance(first_bs, StorePath) and isinstance(first_bs, SyncByteSetter)) ): - self.write_sync(batch, value, drop_axes, max_workers=_resolve_max_workers()) + # One thread hop for the whole batch; see the matching comment in + # `read` for why write_sync must not run inline on the event loop. + await asyncio.to_thread( + self.write_sync, batch, value, drop_axes, max_workers=_resolve_max_workers() + ) return await _async_write_fallback(self, batch, value, drop_axes) diff --git a/tests/benchmarks/test_e2e.py b/tests/benchmarks/test_e2e.py index 9d60d9a2fb..487485e262 100644 --- a/tests/benchmarks/test_e2e.py +++ b/tests/benchmarks/test_e2e.py @@ -190,3 +190,51 @@ def setup() -> tuple[tuple[zarr.Array, EllipsisType], dict]: # type: ignore[typ return (arr, Ellipsis), {} benchmark.pedantic(getitem, setup=setup, rounds=3) # type: ignore[no-untyped-call] + + +_CONCURRENT_READ_THREADS = 8 +_concurrent_layout = Layout(shape=(64_000_000,), chunks=(4_000_000,), shards=None) + + +@pytest.mark.parametrize("pipeline", ["batched", "fused_full_threaded"], indirect=True) +@pytest.mark.parametrize("compression_name", ["zstd", None]) +@pytest.mark.parametrize("store", ["local"], indirect=["store"]) +def test_read_array_concurrent( + bench_store: Store, + compression_name: CompressorName, + pipeline: str, + benchmark: BenchmarkFixture, +) -> None: + """Dask-style access: several user threads each reading one chunk per call. + + All sync-API calls are serviced by the one global event loop, so this + measures how much of each read's IO+compute the pipeline runs while + holding the loop: anything inline serializes the readers. + """ + from concurrent.futures import ThreadPoolExecutor + + layout = _concurrent_layout + arr = create_array( + bench_store, + dtype="uint8", + shape=layout.shape, + chunks=layout.chunks, + shards=layout.shards, + compressors=compressors[compression_name], # type: ignore[arg-type] + fill_value=0, + ) + arr[:] = _data(layout.shape) + selections = [ + slice(start, start + layout.chunks[0]) + for start in range(0, layout.shape[0], layout.chunks[0]) + ] + + def read_all_chunks_concurrently() -> None: + with ThreadPoolExecutor(max_workers=_CONCURRENT_READ_THREADS) as executor: + list(executor.map(lambda sel: arr[sel], selections)) + + def setup() -> tuple[tuple[()], dict]: # type: ignore[type-arg] + clear_cache() + return (), {} + + benchmark.pedantic(read_all_chunks_concurrently, setup=setup, rounds=3) # type: ignore[no-untyped-call] diff --git a/tests/test_fused_pipeline.py b/tests/test_fused_pipeline.py index 73c2c6e1c3..70eaee0398 100644 --- a/tests/test_fused_pipeline.py +++ b/tests/test_fused_pipeline.py @@ -34,6 +34,59 @@ def test_construction(codecs: tuple[Any, ...]) -> None: assert pipeline.codecs == codecs +def test_sync_api_compute_off_event_loop(monkeypatch: pytest.MonkeyPatch) -> None: + """Codec compute must never run on the thread driving the event loop. + + Every sync-API call, from every user thread, is serviced by the one global + `zarr_io` event loop. Running decode/encode inline on that loop's thread + turns it into a mutex around codec compute: concurrent readers serialize, + and the penalty grows with codec cost (observed as "fused pipeline is + slower for zstd data" under dask-style multi-threaded single-chunk reads). + """ + import asyncio + + from zarr.core.chunk_utils import ChunkTransform + + compute_on_loop = {"decode": False, "encode": False} + real_decode = ChunkTransform.decode_chunk + real_encode = ChunkTransform.encode_chunk + + def _running_loop() -> bool: + try: + asyncio.get_running_loop() + except RuntimeError: + return False + return True + + def traced_decode(self: ChunkTransform, chunk_bytes: Any, chunk_spec: Any) -> Any: + compute_on_loop["decode"] = compute_on_loop["decode"] or _running_loop() + return real_decode(self, chunk_bytes, chunk_spec) + + def traced_encode(self: ChunkTransform, chunk_array: Any, chunk_spec: Any) -> Any: + compute_on_loop["encode"] = compute_on_loop["encode"] or _running_loop() + return real_encode(self, chunk_array, chunk_spec) + + monkeypatch.setattr(ChunkTransform, "decode_chunk", traced_decode) + monkeypatch.setattr(ChunkTransform, "encode_chunk", traced_encode) + + with zarr_config.set({"codec_pipeline.path": "zarr.core.codec_pipeline.FusedCodecPipeline"}): + arr = zarr.create_array( + MemoryStore(), + shape=(8, 8), + chunks=(4, 4), + dtype="float64", + compressors=ZstdCodec(level=1), + ) + data = np.arange(64, dtype="float64").reshape(8, 8) + arr[:4, :4] = data[:4, :4] # single-chunk write (batch of 1) + arr[:] = data # multi-chunk write + assert compute_on_loop["encode"] is False + + np.testing.assert_array_equal(arr[:4, :4], data[:4, :4]) # single-chunk read + np.testing.assert_array_equal(arr[:], data) # multi-chunk read + assert compute_on_loop["decode"] is False + + def test_evolve_from_array_spec() -> None: """evolve_from_array_spec creates a sync transform.""" from zarr.core.array_spec import ArrayConfig, ArraySpec From 885523e6c55c53437ac1d8b4eda2e5489ec7e329 Mon Sep 17 00:00:00 2001 From: Davis Vann Bennett Date: Wed, 29 Jul 2026 12:16:49 +0200 Subject: [PATCH 3/4] docs: rename change note to upstream PR number (4194) towncrier's issue_format links to zarr-developers/zarr-python issues, so 247 (the fork PR number) would render a link to an unrelated old issue. Assisted-by: ClaudeCode:claude-fable-5 --- changes/{247.bugfix.md => 4194.bugfix.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename changes/{247.bugfix.md => 4194.bugfix.md} (100%) diff --git a/changes/247.bugfix.md b/changes/4194.bugfix.md similarity index 100% rename from changes/247.bugfix.md rename to changes/4194.bugfix.md From f06bce59c0a42de588675b67d1f7f8eb6b1dc9fa Mon Sep 17 00:00:00 2001 From: Davis Vann Bennett Date: Wed, 29 Jul 2026 12:17:58 +0200 Subject: [PATCH 4/4] test: guard event-loop test against vacuity if the sync fast path stops triggering The test asserts a negative (compute never ran on the loop thread). If a refactor made the sync fast path stop triggering, the traced ChunkTransform methods would never be called (the async fallback uses AsyncChunkTransform) and the test would pass while guarding nothing. Assert the traced hooks actually ran. Assisted-by: ClaudeCode:claude-fable-5 --- tests/test_fused_pipeline.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/test_fused_pipeline.py b/tests/test_fused_pipeline.py index 70eaee0398..09e9241c07 100644 --- a/tests/test_fused_pipeline.py +++ b/tests/test_fused_pipeline.py @@ -48,6 +48,7 @@ def test_sync_api_compute_off_event_loop(monkeypatch: pytest.MonkeyPatch) -> Non from zarr.core.chunk_utils import ChunkTransform compute_on_loop = {"decode": False, "encode": False} + calls = {"decode": 0, "encode": 0} real_decode = ChunkTransform.decode_chunk real_encode = ChunkTransform.encode_chunk @@ -59,10 +60,12 @@ def _running_loop() -> bool: return True def traced_decode(self: ChunkTransform, chunk_bytes: Any, chunk_spec: Any) -> Any: + calls["decode"] += 1 compute_on_loop["decode"] = compute_on_loop["decode"] or _running_loop() return real_decode(self, chunk_bytes, chunk_spec) def traced_encode(self: ChunkTransform, chunk_array: Any, chunk_spec: Any) -> Any: + calls["encode"] += 1 compute_on_loop["encode"] = compute_on_loop["encode"] or _running_loop() return real_encode(self, chunk_array, chunk_spec) @@ -80,10 +83,15 @@ def traced_encode(self: ChunkTransform, chunk_array: Any, chunk_spec: Any) -> An data = np.arange(64, dtype="float64").reshape(8, 8) arr[:4, :4] = data[:4, :4] # single-chunk write (batch of 1) arr[:] = data # multi-chunk write + # Guard against vacuity: if the sync fast path stops triggering, the + # traced ChunkTransform methods are never called (the async fallback + # uses AsyncChunkTransform) and the on-loop flags stay trivially False. + assert calls["encode"] > 0 assert compute_on_loop["encode"] is False np.testing.assert_array_equal(arr[:4, :4], data[:4, :4]) # single-chunk read np.testing.assert_array_equal(arr[:], data) # multi-chunk read + assert calls["decode"] > 0 assert compute_on_loop["decode"] is False