zarr-developers · d-v-b · Mar 31, 2026 · Feb 25, 2026 · Feb 26, 2026 · Feb 26, 2026
diff --git a/src/zarr/abc/codec.py b/src/zarr/abc/codec.py
@@ -67,20 +67,19 @@ def _check_codecjson_v2(data: object) -> TypeGuard[CodecJSON_V2[str]]:
 
 
 @runtime_checkable
-class SupportsSyncCodec(Protocol):
+class SupportsSyncCodec[CI: CodecInput, CO: CodecOutput](Protocol):
     """Protocol for codecs that support synchronous encode/decode.
 
-    Codecs implementing this protocol provide ``_decode_sync`` and ``_encode_sync``
+    Codecs implementing this protocol provide `_decode_sync` and `_encode_sync`
     methods that perform encoding/decoding without requiring an async event loop.
+
+    The type parameters mirror `BaseCodec`: `CI` is the decoded type and `CO` is
+    the encoded type.
     """
 
-    def _decode_sync(
-        self, chunk_data: NDBuffer | Buffer, chunk_spec: ArraySpec
-    ) -> NDBuffer | Buffer: ...
+    def _decode_sync(self, chunk_data: CO, chunk_spec: ArraySpec) -> CI: ...
 
-    def _encode_sync(
-        self, chunk_data: NDBuffer | Buffer, chunk_spec: ArraySpec
-    ) -> NDBuffer | Buffer | None: ...
+    def _encode_sync(self, chunk_data: CI, chunk_spec: ArraySpec) -> CO | None: ...
 
 
 class BaseCodec[CI: CodecInput, CO: CodecOutput](Metadata):

diff --git a/src/zarr/core/codec_pipeline.py b/src/zarr/core/codec_pipeline.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from itertools import islice, pairwise
 from typing import TYPE_CHECKING, Any
 from warnings import warn
@@ -14,6 +14,7 @@
     Codec,
     CodecPipeline,
     GetResult,
+    SupportsSyncCodec,
 )
 from zarr.core.common import concurrent_map
 from zarr.core.config import config
@@ -66,6 +67,111 @@ def fill_value_or_default(chunk_spec: ArraySpec) -> Any:
         return fill_value
 
 
+@dataclass(slots=True, kw_only=True)
+class ChunkTransform:
+    """A synchronous codec chain bound to an ArraySpec.
+
+    Provides `encode` and `decode` for pure-compute codec operations
+    (no IO, no threading, no batching).
+
+    All codecs must implement `SupportsSyncCodec`. Construction will
+    raise `TypeError` if any codec does not.
+    """
+
+    codecs: tuple[Codec, ...]
+    array_spec: ArraySpec
+
+    # (sync codec, input_spec) pairs in pipeline order.
+    _aa_codecs: tuple[tuple[SupportsSyncCodec[NDBuffer, NDBuffer], ArraySpec], ...] = field(
+        init=False, repr=False, compare=False
+    )
+    _ab_codec: SupportsSyncCodec[NDBuffer, Buffer] = field(init=False, repr=False, compare=False)
+    _ab_spec: ArraySpec = field(init=False, repr=False, compare=False)
+    _bb_codecs: tuple[SupportsSyncCodec[Buffer, Buffer], ...] = field(
+        init=False, repr=False, compare=False
+    )
+
+    def __post_init__(self) -> None:
+        non_sync = [c for c in self.codecs if not isinstance(c, SupportsSyncCodec)]
+        if non_sync:
+            names = ", ".join(type(c).__name__ for c in non_sync)
+            raise TypeError(
+                f"All codecs must implement SupportsSyncCodec. The following do not: {names}"
+            )
+
+        aa, ab, bb = codecs_from_list(list(self.codecs))
+
+        aa_codecs: list[tuple[SupportsSyncCodec[NDBuffer, NDBuffer], ArraySpec]] = []
+        spec = self.array_spec
+        for aa_codec in aa:
+            assert isinstance(aa_codec, SupportsSyncCodec)
+            aa_codecs.append((aa_codec, spec))
+            spec = aa_codec.resolve_metadata(spec)
+
+        self._aa_codecs = tuple(aa_codecs)
+        assert isinstance(ab, SupportsSyncCodec)
+        self._ab_codec = ab
+        self._ab_spec = spec
+        bb_sync: list[SupportsSyncCodec[Buffer, Buffer]] = []
+        for bb_codec in bb:
+            assert isinstance(bb_codec, SupportsSyncCodec)
+            bb_sync.append(bb_codec)
+        self._bb_codecs = tuple(bb_sync)
+
+    def decode(
+        self,
+        chunk_bytes: Buffer,
+    ) -> NDBuffer:
+        """Decode a single chunk through the full codec chain, synchronously.
+
+        Pure compute -- no IO.
+        """
+        data: Buffer = chunk_bytes
+        for bb_codec in reversed(self._bb_codecs):
+            data = bb_codec._decode_sync(data, self._ab_spec)
+
+        chunk_array: NDBuffer = self._ab_codec._decode_sync(data, self._ab_spec)
+
+        for aa_codec, spec in reversed(self._aa_codecs):
+            chunk_array = aa_codec._decode_sync(chunk_array, spec)
+
+        return chunk_array
+
+    def encode(
+        self,
+        chunk_array: NDBuffer,
+    ) -> Buffer | None:
+        """Encode a single chunk through the full codec chain, synchronously.
+
+        Pure compute -- no IO.
+        """
+        aa_data: NDBuffer = chunk_array
+        for aa_codec, spec in self._aa_codecs:
+            aa_result = aa_codec._encode_sync(aa_data, spec)
+            if aa_result is None:
+                return None
+            aa_data = aa_result
+
+        ab_result = self._ab_codec._encode_sync(aa_data, self._ab_spec)
+        if ab_result is None:
+            return None
+
+        bb_data: Buffer = ab_result
+        for bb_codec in self._bb_codecs:
+            bb_result = bb_codec._encode_sync(bb_data, self._ab_spec)
+            if bb_result is None:
+                return None
+            bb_data = bb_result
+
+        return bb_data
+
+    def compute_encoded_size(self, byte_length: int, array_spec: ArraySpec) -> int:
+        for codec in self.codecs:
+            byte_length = codec.compute_encoded_size(byte_length, array_spec)
+            array_spec = codec.resolve_metadata(array_spec)
+        return byte_length
+
+
 @dataclass(frozen=True)
 class BatchedCodecPipeline(CodecPipeline):
     """Default codec pipeline.

diff --git a/tests/test_sync_codec_pipeline.py b/tests/test_sync_codec_pipeline.py
@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+import pytest
+
+from zarr.abc.codec import ArrayBytesCodec, Codec
+from zarr.codecs.bytes import BytesCodec
+from zarr.codecs.crc32c_ import Crc32cCodec
+from zarr.codecs.gzip import GzipCodec
+from zarr.codecs.transpose import TransposeCodec
+from zarr.codecs.zstd import ZstdCodec
+from zarr.core.array_spec import ArrayConfig, ArraySpec
+from zarr.core.buffer import Buffer, NDBuffer, default_buffer_prototype
+from zarr.core.codec_pipeline import ChunkTransform
+from zarr.core.dtype import get_data_type_from_native_dtype
+
+
+class AsyncOnlyCodec(ArrayBytesCodec):
+    """A codec that only supports async, for testing rejection of non-sync codecs."""
+
+    is_fixed_size = True
+
+    async def _decode_single(self, chunk_data: Buffer, chunk_spec: ArraySpec) -> NDBuffer:
+        raise NotImplementedError  # pragma: no cover
+
+    async def _encode_single(self, chunk_data: NDBuffer, chunk_spec: ArraySpec) -> Buffer | None:
+        raise NotImplementedError  # pragma: no cover
+
+    def compute_encoded_size(self, input_byte_length: int, chunk_spec: ArraySpec) -> int:
+        return input_byte_length  # pragma: no cover
+
+
+def _make_array_spec(shape: tuple[int, ...], dtype: np.dtype[np.generic]) -> ArraySpec:
+    zdtype = get_data_type_from_native_dtype(dtype)
+    return ArraySpec(
+        shape=shape,
+        dtype=zdtype,
+        fill_value=zdtype.cast_scalar(0),
+        config=ArrayConfig(order="C", write_empty_chunks=True),
+        prototype=default_buffer_prototype(),
+    )
+
+
+def _make_nd_buffer(arr: np.ndarray[Any, np.dtype[Any]]) -> NDBuffer:
+    return default_buffer_prototype().nd_buffer.from_numpy_array(arr)
+
+
+@pytest.mark.parametrize(
+    ("shape", "codecs"),
+    [
+        ((100,), (BytesCodec(),)),
+        ((100,), (BytesCodec(), GzipCodec())),
+        ((3, 4), (TransposeCodec(order=(1, 0)), BytesCodec(), ZstdCodec())),
+    ],
+    ids=["bytes-only", "with-compression", "full-chain"],
+)
+def test_construction(shape: tuple[int, ...], codecs: tuple[Codec, ...]) -> None:
+    """Construction succeeds when all codecs implement SupportsSyncCodec."""
+    spec = _make_array_spec(shape, np.dtype("float64"))
+    ChunkTransform(codecs=codecs, array_spec=spec)
+
+
+@pytest.mark.parametrize(
+    ("shape", "codecs"),
+    [
+        ((100,), (AsyncOnlyCodec(),)),
+        ((3, 4), (TransposeCodec(order=(1, 0)), AsyncOnlyCodec())),
+    ],
+    ids=["async-only", "mixed-sync-and-async"],
+)
+def test_construction_rejects_non_sync(shape: tuple[int, ...], codecs: tuple[Codec, ...]) -> None:
+    """Construction raises TypeError when any codec lacks SupportsSyncCodec."""
+    spec = _make_array_spec(shape, np.dtype("float64"))
+    with pytest.raises(TypeError, match="AsyncOnlyCodec"):
+        ChunkTransform(codecs=codecs, array_spec=spec)
+
+
+@pytest.mark.parametrize(
+    ("arr", "codecs"),
+    [
+        (np.arange(100, dtype="float64"), (BytesCodec(),)),
+        (np.arange(100, dtype="float64"), (BytesCodec(), GzipCodec(level=1))),
+        (
+            np.arange(12, dtype="float64").reshape(3, 4),
+            (TransposeCodec(order=(1, 0)), BytesCodec(), ZstdCodec(level=1)),
+        ),
+        (np.arange(100, dtype="float64"), (BytesCodec(), Crc32cCodec())),
+        (np.arange(50, dtype="int32"), (BytesCodec(), ZstdCodec(level=1))),
+    ],
+    ids=["bytes-only", "gzip", "transpose+zstd", "crc32c", "int32"],
+)
+def test_encode_decode_roundtrip(
+    arr: np.ndarray[Any, np.dtype[Any]], codecs: tuple[Codec, ...]
+) -> None:
+    """Data survives a full encode/decode cycle."""
+    spec = _make_array_spec(arr.shape, arr.dtype)
+    chain = ChunkTransform(codecs=codecs, array_spec=spec)
+    nd_buf = _make_nd_buffer(arr)
+
+    encoded = chain.encode(nd_buf)
+    assert encoded is not None
+    decoded = chain.decode(encoded)
+    np.testing.assert_array_equal(arr, decoded.as_numpy_array())
+
+
+@pytest.mark.parametrize(
+    ("shape", "codecs", "input_size", "expected_size"),
+    [
+        ((100,), (BytesCodec(),), 800, 800),
+        ((100,), (BytesCodec(), Crc32cCodec()), 800, 804),
+        ((3, 4), (TransposeCodec(order=(1, 0)), BytesCodec()), 96, 96),
+    ],
+    ids=["bytes-only", "crc32c", "transpose"],
+)
+def test_compute_encoded_size(
+    shape: tuple[int, ...],
+    codecs: tuple[Codec, ...],
+    input_size: int,
+    expected_size: int,
+) -> None:
+    """compute_encoded_size returns the correct byte length."""
+    spec = _make_array_spec(shape, np.dtype("float64"))
+    chain = ChunkTransform(codecs=codecs, array_spec=spec)
+    assert chain.compute_encoded_size(input_size, spec) == expected_size
+
+
+def test_encode_returns_none_propagation() -> None:
+    """When an AA codec returns None, encode short-circuits and returns None."""
+
+    class NoneReturningAACodec(TransposeCodec):
+        """An ArrayArrayCodec that always returns None from encode."""
+
+        def _encode_sync(self, chunk_array: NDBuffer, chunk_spec: ArraySpec) -> NDBuffer | None:
+            return None
+
+    spec = _make_array_spec((3, 4), np.dtype("float64"))
+    chain = ChunkTransform(
+        codecs=(NoneReturningAACodec(order=(1, 0)), BytesCodec()),
+        array_spec=spec,
+    )
+    arr = np.arange(12, dtype="float64").reshape(3, 4)
+    nd_buf = _make_nd_buffer(arr)
+    assert chain.encode(nd_buf) is None