MRE:
"""Reproducer: FusedCodecPipeline crashes on sharded arrays with async-only inner codecs.
A third-party codec that implements only the async codec interface
(`_encode_single` / `_decode_single`, no `SupportsSyncCodec`) works fine under
the default BatchedCodecPipeline, and under FusedCodecPipeline when used at the
top level of the codec chain. But as an *inner* codec of a sharded array under
FusedCodecPipeline, writing (and reading) raises:
TypeError: All codecs must implement SupportsSyncCodec. The following do not: AsyncOnlyNoop
Run: python repro_fused_async_inner.py
"""
import numpy as np
import zarr
from zarr.abc.codec import BytesBytesCodec
class AsyncOnlyNoop(BytesBytesCodec):
"""No-op bytes-to-bytes codec implementing only the async codec interface."""
is_fixed_size = True
def to_dict(self):
return {"name": "async-only-noop", "configuration": {}}
@classmethod
def from_dict(cls, data):
return cls()
def compute_encoded_size(self, input_byte_length, _spec):
return input_byte_length
async def _encode_single(self, chunk_bytes, chunk_spec):
return chunk_bytes
async def _decode_single(self, chunk_bytes, chunk_spec):
return chunk_bytes
data = np.arange(64, dtype="i4").reshape(8, 8)
for pipeline in ("BatchedCodecPipeline", "FusedCodecPipeline"):
with zarr.config.set({"codec_pipeline.path": f"zarr.core.codec_pipeline.{pipeline}"}):
try:
z = zarr.create_array(
store={},
shape=(8, 8),
chunks=(2, 2),
shards=(4, 4),
dtype="i4",
compressors=(AsyncOnlyNoop(),), # inner codec chain of the shard
)
z[:] = data
ok = np.array_equal(z[:], data)
print(f"{pipeline}: round-trip {'OK' if ok else 'CORRUPTED'}")
except TypeError as e:
print(f"{pipeline}: {type(e).__name__}: {e}")To fix this and raise an error at the correct time (codecpipeline construction time) we need to make some changes to the fused codec pipeline. those will come in a PR.
MRE:
To fix this and raise an error at the correct time (codecpipeline construction time) we need to make some changes to the fused codec pipeline. those will come in a PR.