fix: update tests for simplified pipeline, add n_workers parameter

- Remove test_read_plan.py and test_write_plan.py (tested removed
  layout abstraction)
- Fix test_evolve_from_array_spec to check _sync_transform instead
  of layout
- Replace test_simple_layout_decode_skips_indexer with
  test_sync_transform_encode_decode_roundtrip
- Add n_workers parameter to read_sync/write_sync for thread-pool
  parallelism

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: d-v-b

src/zarr/core/codec_pipeline.py

@@ -938,6 +938,7 @@ def read_sync(
         batch_info: Iterable[tuple[ByteGetter, ArraySpec, SelectorTuple, SelectorTuple, bool]],
         out: NDBuffer,
         drop_axes: tuple[int, ...] = (),
+        n_workers: int = 0,
     ) -> tuple[GetResult, ...]:
         """Synchronous read: fetch -> decode -> scatter, per chunk."""
         assert self._sync_transform is not None
@@ -978,6 +979,7 @@ def write_sync(
         batch_info: Iterable[tuple[ByteSetter, ArraySpec, SelectorTuple, SelectorTuple, bool]],
         value: NDBuffer,
         drop_axes: tuple[int, ...] = (),
+        n_workers: int = 0,
     ) -> None:
         """Synchronous write: merge -> encode -> store, per chunk."""
         assert self._sync_transform is not None

tests/test_phased_codec_pipeline.py

@@ -62,13 +62,13 @@ def test_construction(codecs: tuple[Any, ...]) -> None:
 
 
 def test_evolve_from_array_spec() -> None:
-    """evolve_from_array_spec creates a ChunkLayout."""
+    """evolve_from_array_spec creates a sync transform."""
     from zarr.core.array_spec import ArrayConfig, ArraySpec
     from zarr.core.buffer import default_buffer_prototype
     from zarr.core.dtype import get_data_type_from_native_dtype
 
     pipeline = PhasedCodecPipeline.from_codecs((BytesCodec(),))
-    assert pipeline.layout is None
+    assert pipeline._sync_transform is None
 
     zdtype = get_data_type_from_native_dtype(np.dtype("float64"))
     spec = ArraySpec(
@@ -79,7 +79,7 @@ def test_evolve_from_array_spec() -> None:
         prototype=default_buffer_prototype(),
     )
     evolved = pipeline.evolve_from_array_spec(spec)
-    assert evolved.layout is not None
+    assert evolved._sync_transform is not None
 
 
 @pytest.mark.parametrize(
@@ -294,8 +294,8 @@ async def test_sync_write_async_read_roundtrip() -> None:
     )
 
 
-def test_simple_layout_decode_skips_indexer() -> None:
-    """Non-sharded decode should not create BasicIndexer or ChunkGrid."""
+def test_sync_transform_encode_decode_roundtrip() -> None:
+    """Sync transform can encode and decode a chunk."""
     from zarr.core.array_spec import ArrayConfig, ArraySpec
     from zarr.core.buffer import default_buffer_prototype
     from zarr.core.dtype import Float64
@@ -311,22 +311,17 @@ def test_simple_layout_decode_skips_indexer() -> None:
         config=ArrayConfig(order="C", write_empty_chunks=True),
     )
     pipeline = pipeline.evolve_from_array_spec(spec)
+    assert pipeline._sync_transform is not None
 
-    # Encode some data
+    # Encode
     proto = default_buffer_prototype()
     data = proto.nd_buffer.from_numpy_array(np.arange(100, dtype="float64"))
-    assert pipeline.layout is not None
-    encoded = pipeline.layout.inner_transform.encode_chunk(data)
+    encoded = pipeline._sync_transform.encode_chunk(data)
     assert encoded is not None
 
-    # Decode via decode_chunks_from_index using leaf_transform
-    from zarr.core.codec_pipeline import ShardIndex, decode_chunks_from_index
-
-    index = pipeline.layout.resolve_index(None, "c/0")
-    raw_chunks = {(0,): encoded}
-    result = decode_chunks_from_index(raw_chunks, index, spec)
-    assert result is not None
-    np.testing.assert_array_equal(result.as_numpy_array(), np.arange(100, dtype="float64"))
+    # Decode
+    decoded = pipeline._sync_transform.decode_chunk(encoded)
+    np.testing.assert_array_equal(decoded.as_numpy_array(), np.arange(100, dtype="float64"))
 
 
 # ---------------------------------------------------------------------------