add sync paths for codecpipeline

Author: d-v-b

src/zarr/abc/codec.py

@@ -19,7 +19,7 @@
     from zarr.abc.store import ByteGetter, ByteSetter, Store
     from zarr.core.array_spec import ArraySpec
     from zarr.core.chunk_grids import ChunkGrid
-    from zarr.core.codec_pipeline import ChunkRequest
+    from zarr.core.codec_pipeline import ChunkTransform, ReadChunkRequest, WriteChunkRequest
     from zarr.core.dtype.wrapper import TBaseDType, TBaseScalar, ZDType
     from zarr.core.indexing import ChunkProjection, SelectorTuple
     from zarr.core.metadata import ArrayMetadata
@@ -711,6 +711,20 @@ def compute_encoded_size(self, byte_length: int, array_spec: ArraySpec) -> int:
         """
         ...
 
+    @abstractmethod
+    def get_chunk_transform(self, array_spec: ArraySpec) -> ChunkTransform:
+        """Creates a ChunkTransform for the given array spec.
+
+        Parameters
+        ----------
+        array_spec : ArraySpec
+
+        Returns
+        -------
+        ChunkTransform
+        """
+        ...
+
     @abstractmethod
     async def decode(
         self,
@@ -752,7 +766,7 @@ async def encode(
     @abstractmethod
     async def read(
         self,
-        batch_info: Iterable[ChunkRequest],
+        batch_info: Iterable[ReadChunkRequest],
         out: NDBuffer,
         drop_axes: tuple[int, ...] = (),
     ) -> None:
@@ -761,10 +775,10 @@ async def read(
 
         Parameters
         ----------
-        batch_info : Iterable[ChunkRequest]
-            Ordered set of chunk requests. Each ``ChunkRequest`` carries the
-            store path (``byte_setter``), the ``ArraySpec`` for that chunk,
-            chunk and output selections, and whether the chunk is complete.
+        batch_info : Iterable[ReadChunkRequest]
+            Ordered set of read requests. Each carries a ``byte_getter``,
+            a ``ChunkTransform`` (codec chain + spec), and chunk/output
+            selections.
 
             If the Store returns ``None`` for a chunk, then the chunk was not
             written and the implementation must set the values of that chunk (or
@@ -777,7 +791,7 @@ async def read(
     @abstractmethod
     async def write(
         self,
-        batch_info: Iterable[ChunkRequest],
+        batch_info: Iterable[WriteChunkRequest],
         value: NDBuffer,
         drop_axes: tuple[int, ...] = (),
     ) -> None:
@@ -787,14 +801,37 @@ async def write(
 
         Parameters
         ----------
-        batch_info : Iterable[ChunkRequest]
-            Ordered set of chunk requests. Each ``ChunkRequest`` carries the
-            store path (``byte_setter``), the ``ArraySpec`` for that chunk,
-            chunk and output selections, and whether the chunk is complete.
+        batch_info : Iterable[WriteChunkRequest]
+            Ordered set of write requests. Each carries a ``byte_setter``,
+            a ``ChunkTransform`` (codec chain + spec), chunk/output
+            selections, and whether the chunk is complete.
         value : NDBuffer
         """
         ...
 
+    @property
+    def supports_sync_io(self) -> bool:
+        """Whether this pipeline can run read/write entirely on the calling thread."""
+        return False
+
+    def read_sync(
+        self,
+        batch_info: Iterable[ReadChunkRequest],
+        out: NDBuffer,
+        drop_axes: tuple[int, ...] = (),
+    ) -> None:
+        """Synchronous read: fetch bytes from store, decode, scatter into *out*."""
+        raise NotImplementedError
+
+    def write_sync(
+        self,
+        batch_info: Iterable[WriteChunkRequest],
+        value: NDBuffer,
+        drop_axes: tuple[int, ...] = (),
+    ) -> None:
+        """Synchronous write: gather from *value*, encode, persist to store."""
+        raise NotImplementedError
+
 
 async def _batching_helper(
     func: Callable[[CodecInput, ArraySpec], Awaitable[CodecOutput | None]],

src/zarr/core/array.py

@@ -48,7 +48,7 @@
     V2ChunkKeyEncoding,
     parse_chunk_key_encoding,
 )
-from zarr.core.codec_pipeline import ChunkRequest
+from zarr.core.codec_pipeline import ReadChunkRequest, WriteChunkRequest
 from zarr.core.common import (
     JSON,
     ZARR_JSON,
@@ -5603,14 +5603,15 @@ async def _get_selection(
         # reading chunks and decoding them
         await codec_pipeline.read(
             [
-                ChunkRequest(
-                    byte_setter=store_path / metadata.encode_chunk_key(chunk_coords),
-                    chunk_spec=metadata.get_chunk_spec(chunk_coords, _config, prototype=prototype),
+                ReadChunkRequest(
+                    byte_getter=store_path / metadata.encode_chunk_key(chunk_coords),
+                    transform=codec_pipeline.get_chunk_transform(
+                        metadata.get_chunk_spec(chunk_coords, _config, prototype=prototype)
+                    ),
                     chunk_selection=chunk_selection,
                     out_selection=out_selection,
-                    is_complete_chunk=is_complete_chunk,
                 )
-                for chunk_coords, chunk_selection, out_selection, is_complete_chunk in indexer
+                for chunk_coords, chunk_selection, out_selection, _is_complete in indexer
             ],
             out_buffer,
             drop_axes=indexer.drop_axes,
@@ -5913,9 +5914,11 @@ async def _set_selection(
     # merging with existing data and encoding chunks
     await codec_pipeline.write(
         [
-            ChunkRequest(
+            WriteChunkRequest(
                 byte_setter=store_path / metadata.encode_chunk_key(chunk_coords),
-                chunk_spec=metadata.get_chunk_spec(chunk_coords, _config, prototype),
+                transform=codec_pipeline.get_chunk_transform(
+                    metadata.get_chunk_spec(chunk_coords, _config, prototype)
+                ),
                 chunk_selection=chunk_selection,
                 out_selection=out_selection,
                 is_complete_chunk=is_complete_chunk,