diff --git a/docs/release.rst b/docs/release.rst
index 817bdc4f37..f633aea7cc 100644
--- a/docs/release.rst
+++ b/docs/release.rst
@@ -6,14 +6,20 @@ Release notes
     # to document your changes. On releases it will be
     # re-indented so that it does not show up in the notes.
 
-    .. _unreleased:
+.. _unreleased:
 
-    Unreleased
-    ----------
+Unreleased
+----------
 ..
     # .. warning::
     #    Pre-release! Use :command:`pip install --pre zarr` to evaluate this release.
 
+* Improve Zarr V3 support, adding partial store read/write and storage transformers.
+  Add two features of the [v3 spec](https://zarr-specs.readthedocs.io/en/latest/core/v3.0.html):
+  * storage transformers
+  * `get_partial_values` and `set_partial_values`
+  By :user:`Jonathan Striebel <jstriebel>`; :issue:`1096`.
+
 .. _release_2.13.6:
 
 2.13.6
@@ -44,7 +50,10 @@ Bug fixes
 Appreciation
 ~~~~~~~~~~~~~
 
-Special thanks to Outreachy participants for contributing to most of the maintenance PRs. Please read the blog post summarising the contribution phase and welcoming new Outreachy interns: https://zarr.dev/blog/welcoming-outreachy-2022-interns/
+Special thanks to Outreachy participants for contributing to most of the
+maintenance PRs. Please read the blog post summarising the contribution phase
+and welcoming new Outreachy interns:
+https://zarr.dev/blog/welcoming-outreachy-2022-interns/
 
 
 Enhancements
diff --git a/zarr/_storage/store.py b/zarr/_storage/store.py
index 9e265cf383..4d813b8e05 100644
--- a/zarr/_storage/store.py
+++ b/zarr/_storage/store.py
@@ -1,8 +1,10 @@
 import abc
 import os
+from collections import defaultdict
 from collections.abc import MutableMapping
+from copy import copy
 from string import ascii_letters, digits
-from typing import Any, List, Mapping, Optional, Union
+from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple, Union
 
 from zarr.meta import Metadata2, Metadata3
 from zarr.util import normalize_storage_path
@@ -254,6 +256,82 @@ def __setitem__(self, key, value):
     def __getitem__(self, key):
         """Get a value."""
 
+    @abc.abstractmethod
+    def rmdir(self, path=None):
+        """Remove a data path and all its subkeys and related metadata.
+        Expects a path without the data or meta root prefix."""
+
+    @property
+    def supports_efficient_get_partial_values(self):
+        return False
+
+    def get_partial_values(
+        self,
+        key_ranges: Sequence[Tuple[str, Tuple[int, Optional[int]]]]
+    ) -> List[Union[bytes, memoryview, bytearray]]:
+        """Get multiple partial values.
+        key_ranges can be an iterable of key, range pairs,
+        where a range specifies two integers range_start and range_length
+        as a tuple, (range_start, range_length).
+        range_length may be None to indicate to read until the end.
+        range_start may be negative to start reading range_start bytes
+        from the end of the file.
+        A key may occur multiple times with different ranges.
+        Inserts None for missing keys into the returned list."""
+        results: List[Union[bytes, memoryview, bytearray]] = (
+            [None] * len(key_ranges)  # type: ignore[list-item]
+        )
+        indexed_ranges_by_key: Dict[str, List[Tuple[int, Tuple[int, Optional[int]]]]] = (
+            defaultdict(list)
+        )
+        for i, (key, range_) in enumerate(key_ranges):
+            indexed_ranges_by_key[key].append((i, range_))
+        for key, indexed_ranges in indexed_ranges_by_key.items():
+            try:
+                value = self[key]
+            except KeyError:  # pragma: no cover
+                continue
+            for i, (range_from, range_length) in indexed_ranges:
+                if range_length is None:
+                    results[i] = value[range_from:]
+                else:
+                    results[i] = value[range_from:range_from + range_length]
+        return results
+
+    def supports_efficient_set_partial_values(self):
+        return False
+
+    def set_partial_values(self, key_start_values):
+        """Set multiple partial values.
+        key_start_values can be an iterable of key, start and value triplets
+        as tuples, (key, start, value), where start defines the offset in bytes.
+        A key may occur multiple times with different starts and non-overlapping values.
+        Also, start may only be beyond the current value if other values fill the gap.
+        start may be negative to start writing start bytes from the current
+        end of the file, ending the file with the new value."""
+        unique_keys = set(next(zip(*key_start_values)))
+        values = {}
+        for key in unique_keys:
+            old_value = self.get(key)
+            values[key] = None if old_value is None else bytearray(old_value)
+        for key, start, value in key_start_values:
+            if values[key] is None:
+                assert start == 0
+                values[key] = value
+            else:
+                if start > len(values[key]):  # pragma: no cover
+                    raise ValueError(
+                        f"Cannot set value at start {start}, "
+                        + f"since it is beyond the data at key {key}, "
+                        + f"having length {len(values[key])}."
+                    )
+                if start < 0:
+                    values[key][start:] = value
+                else:
+                    values[key][start:start + len(value)] = value
+        for key, value in values.items():
+            self[key] = value
+
     def clear(self):
         """Remove all items from store."""
         self.erase_prefix("/")
@@ -303,6 +381,151 @@ def _ensure_store(store):
         )
 
 
+class StorageTransformer(MutableMapping, abc.ABC):
+    """Base class for storage transformers. The methods simply pass on the data as-is
+    and should be overwritten by sub-classes."""
+
+    _store_version = 3
+    _metadata_class = Metadata3
+
+    def __init__(self, _type) -> None:
+        if _type not in self.valid_types:  # pragma: no cover
+            raise ValueError(
+                f"Storage transformer cannot be initialized with type {_type}, "
+                + f"must be one of {list(self.valid_types)}."
+            )
+        self.type = _type
+        self._inner_store = None
+
+    def _copy_for_array(self, array, inner_store):
+        transformer_copy = copy(self)
+        transformer_copy._inner_store = inner_store
+        return transformer_copy
+
+    @abc.abstractproperty
+    def extension_uri(self):
+        pass  # pragma: no cover
+
+    @abc.abstractproperty
+    def valid_types(self):
+        pass  # pragma: no cover
+
+    def get_config(self):
+        """Return a dictionary holding configuration parameters for this
+        storage transformer. All values must be compatible with JSON encoding."""
+        # Override in sub-class if need special encoding of config values.
+        # By default, assume all non-private members are configuration
+        # parameters except for type .
+        return {
+            k: v for k, v in self.__dict__.items()
+            if not k.startswith('_') and k != "type"
+        }
+
+    @classmethod
+    def from_config(cls, _type, config):
+        """Instantiate storage transformer from a configuration object."""
+        # override in sub-class if need special decoding of config values
+
+        # by default, assume constructor accepts configuration parameters as
+        # keyword arguments without any special decoding
+        return cls(_type, **config)
+
+    @property
+    def inner_store(self) -> Union["StorageTransformer", StoreV3]:
+        assert self._inner_store is not None, (
+            "inner_store is not initialized, first get a copy via _copy_for_array."
+        )
+        return self._inner_store
+
+    # The following implementations are usually fine to keep as-is:
+
+    def __eq__(self, other):
+        return (
+            type(self) == type(other) and
+            self._inner_store == other._inner_store and
+            self.get_config() == other.get_config()
+        )
+
+    def erase(self, key):
+        self.__delitem__(key)
+
+    def list(self):
+        return list(self.keys())
+
+    def list_dir(self, prefix):
+        return StoreV3.list_dir(self, prefix)
+
+    def is_readable(self):
+        return self.inner_store.is_readable()
+
+    def is_writeable(self):
+        return self.inner_store.is_writeable()
+
+    def is_listable(self):
+        return self.inner_store.is_listable()
+
+    def is_erasable(self):
+        return self.inner_store.is_erasable()
+
+    def clear(self):
+        return self.inner_store.clear()
+
+    def __enter__(self):
+        return self.inner_store.__enter__()
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        return self.inner_store.__exit__(exc_type, exc_value, traceback)
+
+    def close(self) -> None:
+        return self.inner_store.close()
+
+    # The following implementations might need to be re-implemented
+    # by subclasses implementing storage transformers:
+
+    def rename(self, src_path: str, dst_path: str) -> None:
+        return self.inner_store.rename(src_path, dst_path)
+
+    def list_prefix(self, prefix):
+        return self.inner_store.list_prefix(prefix)
+
+    def erase_prefix(self, prefix):
+        return self.inner_store.erase_prefix(prefix)
+
+    def rmdir(self, path=None):
+        return self.inner_store.rmdir(path)
+
+    def __contains__(self, key):
+        return self.inner_store.__contains__(key)
+
+    def __setitem__(self, key, value):
+        return self.inner_store.__setitem__(key, value)
+
+    def __getitem__(self, key):
+        return self.inner_store.__getitem__(key)
+
+    def __delitem__(self, key):
+        return self.inner_store.__delitem__(key)
+
+    def __iter__(self):
+        return self.inner_store.__iter__()
+
+    def __len__(self):
+        return self.inner_store.__len__()
+
+    @property
+    def supports_efficient_get_partial_values(self):
+        return self.inner_store.supports_efficient_get_partial_values
+
+    def get_partial_values(self, key_ranges):
+        return self.inner_store.get_partial_values(key_ranges)
+
+    def supports_efficient_set_partial_values(self):
+        return self.inner_store.supports_efficient_set_partial_values()
+
+    def set_partial_values(self, key_start_values):
+        return self.inner_store.set_partial_values(key_start_values)
+
+
 # allow MutableMapping for backwards compatibility
 StoreLike = Union[BaseStore, MutableMapping]
 
diff --git a/zarr/core.py b/zarr/core.py
index e5b2045160..5d37570831 100644
--- a/zarr/core.py
+++ b/zarr/core.py
@@ -189,6 +189,7 @@ def __init__(
 
         self._store = store
         self._chunk_store = chunk_store
+        self._transformed_chunk_store = None
         self._path = normalize_storage_path(path)
         if self._path:
             self._key_prefix = self._path + '/'
@@ -292,6 +293,16 @@ def _load_metadata_nosync(self):
                 filters = [get_codec(config) for config in filters]
             self._filters = filters
 
+            if self._version == 3:
+                storage_transformers = meta.get('storage_transformers', [])
+                if storage_transformers:
+                    transformed_store = self._chunk_store or self._store
+                    for storage_transformer in storage_transformers[::-1]:
+                        transformed_store = storage_transformer._copy_for_array(
+                            self, transformed_store
+                        )
+                    self._transformed_chunk_store = transformed_store
+
     def _refresh_metadata(self):
         if not self._cache_metadata:
             self._load_metadata()
@@ -371,10 +382,12 @@ def read_only(self, value):
     @property
     def chunk_store(self):
         """A MutableMapping providing the underlying storage for array chunks."""
-        if self._chunk_store is None:
-            return self._store
-        else:
+        if self._transformed_chunk_store is not None:
+            return self._transformed_chunk_store
+        elif self._chunk_store is not None:
             return self._chunk_store
+        else:
+            return self._store
 
     @property
     def shape(self):
@@ -1800,7 +1813,7 @@ def _set_selection(self, indexer, value, fields=None):
             check_array_shape('value', value, sel_shape)
 
         # iterate over chunks in range
-        if not hasattr(self.store, "setitems") or self._synchronizer is not None \
+        if not hasattr(self.chunk_store, "setitems") or self._synchronizer is not None \
            or any(map(lambda x: x == 0, self.shape)):
             # iterative approach
             for chunk_coords, chunk_selection, out_selection in indexer:
@@ -2229,7 +2242,10 @@ def _encode_chunk(self, chunk):
             cdata = chunk
 
         # ensure in-memory data is immutable and easy to compare
-        if isinstance(self.chunk_store, KVStore):
+        if (
+            isinstance(self.chunk_store, KVStore)
+            or isinstance(self._chunk_store, KVStore)
+        ):
             cdata = ensure_bytes(cdata)
 
         return cdata
diff --git a/zarr/creation.py b/zarr/creation.py
index cc191e3734..a6fa8e44cc 100644
--- a/zarr/creation.py
+++ b/zarr/creation.py
@@ -22,7 +22,7 @@ def create(shape, chunks=True, dtype=None, compressor='default',
            overwrite=False, path=None, chunk_store=None, filters=None,
            cache_metadata=True, cache_attrs=True, read_only=False,
            object_codec=None, dimension_separator=None, write_empty_chunks=True,
-           *, zarr_version=None, meta_array=None, **kwargs):
+           *, zarr_version=None, meta_array=None, storage_transformers=(), **kwargs):
     """Create an array.
 
     Parameters
@@ -85,6 +85,14 @@ def create(shape, chunks=True, dtype=None, compressor='default',
 
         .. versionadded:: 2.11
 
+    storage_transformers : sequence of StorageTransformers, optional
+        Setting storage transformers, changes the storage structure and behaviour
+        of data coming from the underlying store. The transformers are applied in the
+        order of the given sequence. Supplying an empty sequence is the same as omitting
+        the argument or setting it to None. May only be set when using zarr_version 3.
+
+        .. versionadded:: 2.13
+
     zarr_version : {None, 2, 3}, optional
         The zarr protocol version of the created array. If None, it will be
         inferred from ``store`` or ``chunk_store`` if they are provided,
@@ -170,7 +178,7 @@ def create(shape, chunks=True, dtype=None, compressor='default',
     init_array(store, shape=shape, chunks=chunks, dtype=dtype, compressor=compressor,
                fill_value=fill_value, order=order, overwrite=overwrite, path=path,
                chunk_store=chunk_store, filters=filters, object_codec=object_codec,
-               dimension_separator=dimension_separator)
+               dimension_separator=dimension_separator, storage_transformers=storage_transformers)
 
     # instantiate array
     z = Array(store, path=path, chunk_store=chunk_store, synchronizer=synchronizer,
diff --git a/zarr/meta.py b/zarr/meta.py
index 77c55b9871..41a90101b5 100644
--- a/zarr/meta.py
+++ b/zarr/meta.py
@@ -9,7 +9,11 @@
 from zarr.errors import MetadataError
 from zarr.util import json_dumps, json_loads
 
-from typing import cast, Union, Any, List, Mapping as MappingType, Optional
+from typing import cast, Union, Any, List, Mapping as MappingType, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:  # pragma: no cover
+    from zarr._storage.store import StorageTransformer
+
 
 ZARR_FORMAT = 2
 ZARR_FORMAT_v3 = 3
@@ -459,6 +463,36 @@ def _decode_codec_metadata(cls, meta: Optional[Mapping]) -> Optional[Codec]:
 
         return codec
 
+    @classmethod
+    def _encode_storage_transformer_metadata(
+        cls,
+        storage_transformer: "StorageTransformer"
+    ) -> Optional[Mapping]:
+        return {
+            "extension": storage_transformer.extension_uri,
+            "type": storage_transformer.type,
+            "configuration": storage_transformer.get_config(),
+        }
+
+    @classmethod
+    def _decode_storage_transformer_metadata(cls, meta: Mapping) -> "StorageTransformer":
+        from zarr.tests.test_storage_v3 import DummyStorageTransfomer
+
+        # This might be changed to a proper registry in the future
+        KNOWN_STORAGE_TRANSFORMERS = [DummyStorageTransfomer]
+
+        conf = meta.get('configuration', {})
+        extension_uri = meta['extension']
+        transformer_type = meta['type']
+
+        for StorageTransformerCls in KNOWN_STORAGE_TRANSFORMERS:
+            if StorageTransformerCls.extension_uri == extension_uri:
+                break
+        else:  # pragma: no cover
+            raise NotImplementedError
+
+        return StorageTransformerCls.from_config(transformer_type, conf)
+
     @classmethod
     def decode_array_metadata(cls, s: Union[MappingType, str]) -> MappingType[str, Any]:
         meta = cls.parse_metadata(s)
@@ -476,6 +510,10 @@ def decode_array_metadata(cls, s: Union[MappingType, str]) -> MappingType[str, A
             # TODO: remove dimension_separator?
 
             compressor = cls._decode_codec_metadata(meta.get("compressor", None))
+            storage_transformers = meta.get("storage_transformers", ())
+            storage_transformers = [
+                cls._decode_storage_transformer_metadata(i) for i in storage_transformers
+            ]
             extensions = meta.get("extensions", [])
             meta = dict(
                 shape=tuple(meta["shape"]),
@@ -493,6 +531,8 @@ def decode_array_metadata(cls, s: Union[MappingType, str]) -> MappingType[str, A
             # compressor field should be absent when there is no compression
             if compressor:
                 meta['compressor'] = compressor
+            if storage_transformers:
+                meta['storage_transformers'] = storage_transformers
 
         except Exception as e:
             raise MetadataError("error decoding metadata: %s" % e)
@@ -514,6 +554,10 @@ def encode_array_metadata(cls, meta: MappingType[str, Any]) -> bytes:
             object_codec = None
 
         compressor = cls._encode_codec_metadata(meta.get("compressor", None))
+        storage_transformers = meta.get("storage_transformers", ())
+        storage_transformers = [
+            cls._encode_storage_transformer_metadata(i) for i in storage_transformers
+        ]
         extensions = meta.get("extensions", [])
         meta = dict(
             shape=meta["shape"] + sdshape,
@@ -532,6 +576,8 @@ def encode_array_metadata(cls, meta: MappingType[str, Any]) -> bytes:
             meta["compressor"] = compressor
         if dimension_separator:
             meta["dimension_separator"] = dimension_separator
+        if storage_transformers:
+            meta["storage_transformers"] = storage_transformers
         return json_dumps(meta)
 
 
diff --git a/zarr/storage.py b/zarr/storage.py
index a2a8919d0b..db51cca947 100644
--- a/zarr/storage.py
+++ b/zarr/storage.py
@@ -311,6 +311,7 @@ def init_array(
     filters=None,
     object_codec=None,
     dimension_separator=None,
+    storage_transformers=(),
 ):
     """Initialize an array store with the given configuration. Note that this is a low-level
     function and there should be no need to call this directly from user code.
@@ -438,7 +439,8 @@ def init_array(
                          order=order, overwrite=overwrite, path=path,
                          chunk_store=chunk_store, filters=filters,
                          object_codec=object_codec,
-                         dimension_separator=dimension_separator)
+                         dimension_separator=dimension_separator,
+                         storage_transformers=storage_transformers)
 
 
 def _init_array_metadata(
@@ -455,6 +457,7 @@ def _init_array_metadata(
     filters=None,
     object_codec=None,
     dimension_separator=None,
+    storage_transformers=(),
 ):
 
     store_version = getattr(store, '_store_version', 2)
@@ -576,6 +579,7 @@ def _init_array_metadata(
     if store_version < 3:
         meta.update(dict(chunks=chunks, dtype=dtype, order=order,
                          filters=filters_config))
+        assert not storage_transformers
     else:
         if dimension_separator is None:
             dimension_separator = "/"
@@ -589,7 +593,8 @@ def _init_array_metadata(
                                  separator=dimension_separator),
                  chunk_memory_layout=order,
                  data_type=dtype,
-                 attributes=attributes)
+                 attributes=attributes,
+                 storage_transformers=storage_transformers)
         )
 
     key = _prefix_to_array_key(store, _path_to_prefix(path))
diff --git a/zarr/tests/test_core.py b/zarr/tests/test_core.py
index e32026e662..ffacefb937 100644
--- a/zarr/tests/test_core.py
+++ b/zarr/tests/test_core.py
@@ -49,9 +49,11 @@
     KVStoreV3,
     LMDBStoreV3,
     LRUStoreCacheV3,
+    RmdirV3,
     SQLiteStoreV3,
     StoreV3,
 )
+from zarr.tests.test_storage_v3 import DummyStorageTransfomer
 from zarr.util import buffer_size
 from zarr.tests.util import abs_container, skip_test_env_var, have_fsspec, mktemp
 
@@ -3098,7 +3100,7 @@ def test_nbytes_stored(self):
 # Note: this custom mapping doesn't actually have all methods in the
 #       v3 spec (e.g. erase), but they aren't needed here.
 
-class CustomMappingV3(StoreV3):
+class CustomMappingV3(RmdirV3, StoreV3):
 
     def __init__(self):
         self.inner = KVStoreV3(dict())
@@ -3359,6 +3361,36 @@ def expected(self):
         ]
 
 
+@pytest.mark.skipif(not v3_api_available, reason="V3 is disabled")
+class TestArrayWithStorageTransformersV3(TestArrayWithChunkStoreV3):
+
+    @staticmethod
+    def create_array(array_path='arr1', read_only=False, **kwargs):
+        store = KVStoreV3(dict())
+        # separate chunk store
+        chunk_store = KVStoreV3(dict())
+        cache_metadata = kwargs.pop('cache_metadata', True)
+        cache_attrs = kwargs.pop('cache_attrs', True)
+        write_empty_chunks = kwargs.pop('write_empty_chunks', True)
+        dummy_storage_transformer = DummyStorageTransfomer(
+            "dummy_type", test_value=DummyStorageTransfomer.TEST_CONSTANT
+        )
+        init_array(store, path=array_path, chunk_store=chunk_store,
+                   storage_transformers=[dummy_storage_transformer], **kwargs)
+        return Array(store, path=array_path, read_only=read_only,
+                     chunk_store=chunk_store, cache_metadata=cache_metadata,
+                     cache_attrs=cache_attrs, write_empty_chunks=write_empty_chunks)
+
+    def expected(self):
+        return [
+            "3fb9a4f8233b09ad02067b6b7fc9fd5caa405c7d",
+            "89c8eb364beb84919fc9153d2c1ed2696274ec18",
+            "73307055c3aec095dd1232c38d793ef82a06bd97",
+            "6152c09255a5efa43b1a115546e35affa00c138c",
+            "2f8802fc391f67f713302e84fad4fd8f1366d6c2",
+        ]
+
+
 @pytest.mark.skipif(not v3_api_available, reason="V3 is disabled")
 def test_array_mismatched_store_versions():
     store_v3 = KVStoreV3(dict())
diff --git a/zarr/tests/test_creation.py b/zarr/tests/test_creation.py
index 4c9c292734..b791bc3952 100644
--- a/zarr/tests/test_creation.py
+++ b/zarr/tests/test_creation.py
@@ -19,8 +19,10 @@
 from zarr._storage.store import v3_api_available
 from zarr._storage.v3 import DirectoryStoreV3, KVStoreV3
 from zarr.sync import ThreadSynchronizer
+from zarr.tests.test_storage_v3 import DummyStorageTransfomer
 from zarr.tests.util import mktemp, have_fsspec
 
+
 _VERSIONS = ((None, 2, 3) if v3_api_available else (None, 2))
 _VERSIONS2 = ((2, 3) if v3_api_available else (2, ))
 
@@ -747,3 +749,16 @@ def test_create_read_only(zarr_version, at_root):
 def test_json_dumps_chunks_numpy_dtype():
     z = zeros((10,), chunks=(np.int64(2),))
     assert np.all(z[...] == 0)
+
+
+@pytest.mark.skipif(not v3_api_available, reason="V3 is disabled")
+@pytest.mark.parametrize('at_root', [False, True])
+def test_create_with_storage_transformers(at_root):
+    kwargs = _init_creation_kwargs(zarr_version=3, at_root=at_root)
+    transformer = DummyStorageTransfomer(
+        "dummy_type",
+        test_value=DummyStorageTransfomer.TEST_CONSTANT
+    )
+    z = create(1000000000, chunks=True, storage_transformers=[transformer], **kwargs)
+    assert isinstance(z.chunk_store, DummyStorageTransfomer)
+    assert z.chunk_store.test_value == DummyStorageTransfomer.TEST_CONSTANT
diff --git a/zarr/tests/test_storage_v3.py b/zarr/tests/test_storage_v3.py
index 4f6215135c..9f18c89361 100644
--- a/zarr/tests/test_storage_v3.py
+++ b/zarr/tests/test_storage_v3.py
@@ -1,6 +1,7 @@
 import array
 import atexit
 import copy
+import inspect
 import os
 import tempfile
 
@@ -8,7 +9,7 @@
 import pytest
 
 import zarr
-from zarr._storage.store import _get_hierarchy_metadata, v3_api_available
+from zarr._storage.store import _get_hierarchy_metadata, v3_api_available, StorageTransformer
 from zarr.meta import _default_entry_point_metadata_v3
 from zarr.storage import (atexit_rmglob, atexit_rmtree, data_root,
                           default_compressor, getsize, init_array, meta_root,
@@ -88,6 +89,18 @@ def keys(self):
         """keys"""
 
 
+class DummyStorageTransfomer(StorageTransformer):
+    TEST_CONSTANT = "test1234"
+
+    extension_uri = "https://purl.org/zarr/spec/storage_transformers/dummy/1.0"
+    valid_types = ["dummy_type"]
+
+    def __init__(self, _type, test_value) -> None:
+        super().__init__(_type)
+        assert test_value == self.TEST_CONSTANT
+        self.test_value = test_value
+
+
 def test_ensure_store_v3():
     class InvalidStore:
         pass
@@ -190,8 +203,11 @@ def test_init_array(self, dimension_separator_fixture_v3):
 
         store = self.create_store()
         path = 'arr1'
+        transformer = DummyStorageTransfomer(
+            "dummy_type", test_value=DummyStorageTransfomer.TEST_CONSTANT
+        )
         init_array(store, path=path, shape=1000, chunks=100,
-                   dimension_separator=pass_dim_sep)
+                   dimension_separator=pass_dim_sep, storage_transformers=[transformer])
 
         # check metadata
         mkey = meta_root + path + '.array.json'
@@ -204,6 +220,9 @@ def test_init_array(self, dimension_separator_fixture_v3):
         assert meta['fill_value'] is None
         # Missing MUST be assumed to be "/"
         assert meta['chunk_grid']['separator'] is want_dim_sep
+        assert len(meta["storage_transformers"]) == 1
+        assert isinstance(meta["storage_transformers"][0], DummyStorageTransfomer)
+        assert meta["storage_transformers"][0].test_value == DummyStorageTransfomer.TEST_CONSTANT
         store.close()
 
     def test_list_prefix(self):
@@ -235,6 +254,67 @@ def test_rename_nonexisting(self):
             with pytest.raises(NotImplementedError):
                 store.rename('a', 'b')
 
+    def test_get_partial_values(self):
+        store = self.create_store()
+        store.supports_efficient_get_partial_values in [True, False]
+        store[data_root + 'foo'] = b'abcdefg'
+        store[data_root + 'baz'] = b'z'
+        assert [b'a'] == store.get_partial_values(
+            [
+                (data_root + 'foo', (0, 1))
+            ]
+        )
+        assert [
+            b'd', b'b', b'z', b'abc', b'defg', b'defg', b'g', b'ef'
+        ] == store.get_partial_values(
+            [
+                (data_root + 'foo', (3, 1)),
+                (data_root + 'foo', (1, 1)),
+                (data_root + 'baz', (0, 1)),
+                (data_root + 'foo', (0, 3)),
+                (data_root + 'foo', (3, 4)),
+                (data_root + 'foo', (3, None)),
+                (data_root + 'foo', (-1, None)),
+                (data_root + 'foo', (-3, 2)),
+            ]
+        )
+
+    def test_set_partial_values(self):
+        store = self.create_store()
+        store.supports_efficient_set_partial_values()
+        store[data_root + 'foo'] = b'abcdefg'
+        store.set_partial_values(
+            [
+                (data_root + 'foo', 0, b'hey')
+            ]
+        )
+        assert store[data_root + 'foo'] == b'heydefg'
+
+        store.set_partial_values(
+            [
+                (data_root + 'baz', 0, b'z')
+            ]
+        )
+        assert store[data_root + 'baz'] == b'z'
+        store.set_partial_values(
+            [
+                (data_root + 'foo', 1, b'oo'),
+                (data_root + 'baz', 1, b'zzz'),
+                (data_root + 'baz', 4, b'aaaa'),
+                (data_root + 'foo', 6, b'done'),
+            ]
+        )
+        assert store[data_root + 'foo'] == b'hoodefdone'
+        assert store[data_root + 'baz'] == b'zzzzaaaa'
+        store.set_partial_values(
+            [
+                (data_root + 'foo', -2, b'NE'),
+                (data_root + 'baz', -5, b'q'),
+            ]
+        )
+        assert store[data_root + 'foo'] == b'hoodefdoNE'
+        assert store[data_root + 'baz'] == b'zzzq'
+
 
 class TestMappingStoreV3(StoreV3Tests):
 
@@ -443,6 +523,31 @@ def create_store(self, **kwargs):
         return store
 
 
+class TestStorageTransformerV3(TestMappingStoreV3):
+
+    def create_store(self, **kwargs):
+        inner_store = super().create_store(**kwargs)
+        storage_transformer = DummyStorageTransfomer(
+            "dummy_type", test_value=DummyStorageTransfomer.TEST_CONSTANT
+        )
+        return storage_transformer._copy_for_array(None, inner_store)
+
+    def test_method_forwarding(self):
+        store = self.create_store()
+        assert store.list() == store.inner_store.list()
+        assert store.list_dir(data_root) == store.inner_store.list_dir(data_root)
+
+        assert store.is_readable()
+        assert store.is_writeable()
+        assert store.is_listable()
+        store.inner_store._readable = False
+        store.inner_store._writeable = False
+        store.inner_store._listable = False
+        assert not store.is_readable()
+        assert not store.is_writeable()
+        assert not store.is_listable()
+
+
 class TestLRUStoreCacheV3(_TestLRUStoreCache, StoreV3Tests):
 
     CountingClass = CountingDictV3
@@ -535,3 +640,19 @@ def test_top_level_imports():
             assert hasattr(zarr, store_name)  # pragma: no cover
         else:
             assert not hasattr(zarr, store_name)  # pragma: no cover
+
+
+def _get_public_and_dunder_methods(some_class):
+    return set(
+        name for name, _ in inspect.getmembers(some_class, predicate=inspect.isfunction)
+        if not name.startswith("_") or name.startswith("__")
+    )
+
+
+def test_storage_transformer_interface():
+    store_v3_methods = _get_public_and_dunder_methods(StoreV3)
+    store_v3_methods.discard("__init__")
+    storage_transformer_methods = _get_public_and_dunder_methods(StorageTransformer)
+    storage_transformer_methods.discard("__init__")
+    storage_transformer_methods.discard("get_config")
+    assert storage_transformer_methods == store_v3_methods