fix(codecs): don't skip outer AA/BB codecs in FusedCodecPipeline partial IO

changes/225.bugfix.md

@@ -0,0 +1,16 @@
+`FusedCodecPipeline` no longer skips array->array and bytes->bytes codecs that
+wrap a codec capable of partial IO, such as a `TransposeCodec` around a
+`ShardingCodec`. Partial IO delegates the whole chunk to the array->bytes codec,
+which is only correct when that codec is the entire pipeline, so the pipeline
+now declines partial IO when outer codecs are present — matching
+`BatchedCodecPipeline`.
+
+Previously the outer codec was dropped on both encode and decode. Because the
+pipeline was self-consistent, a roundtrip under `FusedCodecPipeline` alone
+looked correct while the stored bytes did not match the array metadata, so
+`BatchedCodecPipeline` and other Zarr implementations read such arrays back
+incorrectly.
+
+Arrays created by `create_array` are unaffected: `filters`, `order` and
+`compressors` are nested inside the shard, so partial IO remains enabled for
+them.

src/zarr/core/codec_pipeline.py

@@ -132,23 +132,18 @@ def pipeline_supports_partial_decode(
     *,
     array_array_codecs: tuple[ArrayArrayCodec, ...],
     bytes_bytes_codecs: tuple[BytesBytesCodec, ...],
-    require_no_aa_bb: bool,
 ) -> bool:
     """Whether a codec pipeline can decode a partial selection without a full read.
 
     Requires the array->bytes codec to implement
-    ``ArrayBytesCodecPartialDecodeMixin``. When ``require_no_aa_bb`` is True it
-    additionally requires no array->array / bytes->bytes codecs, because those
-    can change the slice<->byte-range correspondence (an AA codec can make the
-    selection non-contiguous, a BB codec can rewrite the bytes), making partial
-    decode infeasible.
-
-    NOTE: the two pipelines currently pass different ``require_no_aa_bb`` values
-    (Batched: True; Fused: False). That divergence is intentional-for-now and
-    tracked separately; this function centralizes the predicate without changing
-    either pipeline's behavior.
+    `ArrayBytesCodecPartialDecodeMixin`, and requires no array->array /
+    bytes->bytes codecs. Partial IO delegates the whole chunk to the
+    array->bytes codec, so it is only sound when that codec is the entire
+    pipeline: an outer AA codec can make the selection non-contiguous and a BB
+    codec can rewrite the bytes, so neither the slice<->byte-range
+    correspondence nor the outer codec itself survives the delegation.
     """
-    if require_no_aa_bb and (len(array_array_codecs) + len(bytes_bytes_codecs)) != 0:
+    if (len(array_array_codecs) + len(bytes_bytes_codecs)) != 0:
         return False
     return isinstance(array_bytes_codec, ArrayBytesCodecPartialDecodeMixin)
 
@@ -158,14 +153,12 @@ def pipeline_supports_partial_encode(
     *,
     array_array_codecs: tuple[ArrayArrayCodec, ...],
     bytes_bytes_codecs: tuple[BytesBytesCodec, ...],
-    require_no_aa_bb: bool,
 ) -> bool:
     """Whether a codec pipeline can encode a partial selection without a full rewrite.
 
-    Mirror of ``pipeline_supports_partial_decode`` for encoding. See its note re:
-    the per-pipeline ``require_no_aa_bb`` divergence.
+    Mirror of `pipeline_supports_partial_decode` for encoding.
     """
-    if require_no_aa_bb and (len(array_array_codecs) + len(bytes_bytes_codecs)) != 0:
+    if (len(array_array_codecs) + len(bytes_bytes_codecs)) != 0:
         return False
     return isinstance(array_bytes_codec, ArrayBytesCodecPartialEncodeMixin)
 
@@ -554,7 +547,6 @@ def supports_partial_decode(self) -> bool:
             self.array_bytes_codec,
             array_array_codecs=self.array_array_codecs,
             bytes_bytes_codecs=self.bytes_bytes_codecs,
-            require_no_aa_bb=True,
         )
 
     @property
@@ -563,7 +555,6 @@ def supports_partial_encode(self) -> bool:
             self.array_bytes_codec,
             array_array_codecs=self.array_array_codecs,
             bytes_bytes_codecs=self.bytes_bytes_codecs,
-            require_no_aa_bb=True,
         )
 
     def __iter__(self) -> Iterator[Codec]:
@@ -934,14 +925,12 @@ def __iter__(self) -> Iterator[Codec]:
 
     @property
     def supports_partial_decode(self) -> bool:
-        # NOTE: unlike BatchedCodecPipeline this does NOT require the AA/BB codec
-        # lists to be empty (require_no_aa_bb=False). That divergence is tracked
-        # separately; see pipeline_supports_partial_decode.
+        # Only a single ArrayBytesCodec that supports partial decoding, and no
+        # AA/BB codecs (they break the slice<->byte-range correspondence).
         return pipeline_supports_partial_decode(
             self.array_bytes_codec,
             array_array_codecs=self.array_array_codecs,
             bytes_bytes_codecs=self.bytes_bytes_codecs,
-            require_no_aa_bb=False,
         )
 
     @property
@@ -950,7 +939,6 @@ def supports_partial_encode(self) -> bool:
             self.array_bytes_codec,
             array_array_codecs=self.array_array_codecs,
             bytes_bytes_codecs=self.bytes_bytes_codecs,
-            require_no_aa_bb=False,
         )
 
     def validate(

tests/test_pipeline_parity.py

@@ -48,6 +48,7 @@
     ShardingCodec,
     SubchunkWriteOrder,
 )
+from zarr.codecs.transpose import TransposeCodec
 from zarr.core.config import config as zarr_config
 from zarr.storage import MemoryStore
 
@@ -332,6 +333,50 @@ def test_pipeline_parity(
     )
 
 
+# ---------------------------------------------------------------------------
+# Outer array->array codec wrapping a sharding codec
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.filterwarnings("ignore:Combining a `sharding_indexed` codec:UserWarning")
+@pytest.mark.parametrize("write_pipeline", [_BATCHED, _FUSED])
+@pytest.mark.parametrize("read_pipeline", [_BATCHED, _FUSED])
+def test_outer_array_array_codec_around_sharding(write_pipeline: str, read_pipeline: str) -> None:
+    """An array->array codec *outside* a sharding codec must not be skipped.
+
+    Partial IO hands the chunk straight to the array->bytes codec, which is only
+    sound when that codec is the whole pipeline. With an outer transpose the
+    partial paths must be declined, or the transpose is silently dropped.
+
+    `filters` are applied outside the serializer, so `create_array` reaches this
+    layout — no low-level `codecs=` needed. Every write/read pipeline pairing is
+    checked because each pipeline is self-consistent on its own: it drops the
+    transpose symmetrically on write and read, so the corruption is only visible
+    when the data crosses pipelines (or implementations).
+    """
+    shape, shard_shape, inner_chunk = (20, 20), (10, 10), (5, 5)
+    ref = np.arange(int(np.prod(shape)), dtype="int32").reshape(shape)
+
+    store = MemoryStore()
+    with zarr_config.set({"codec_pipeline.path": write_pipeline}):
+        arr = zarr.create_array(
+            store=store,
+            shape=shape,
+            chunks=shard_shape,
+            dtype="int32",
+            filters=[TransposeCodec(order=(1, 0))],
+            serializer=ShardingCodec(chunk_shape=inner_chunk),
+            compressors=None,
+        )
+        arr[:] = ref
+
+    np.testing.assert_array_equal(
+        _read_under_pipeline(read_pipeline, store),
+        ref,
+        err_msg=f"outer TransposeCodec dropped (write={write_pipeline}, read={read_pipeline})",
+    )
+
+
 # ---------------------------------------------------------------------------
 # Partial-read parity across subchunk write orders
 # ---------------------------------------------------------------------------