Repository: pangeo-forge/pangeo-forge-recipes (language: python)

Issue:
Concatenate across multiple dimensions
Currently we only support sequential inputs, along a single dimension, commonly time.

```
foo.time0.nc
foo.time1.nc
...
foo.timeN.nc
```

However, many potential recipes instead multiple axes in the files names, e.g. time and ensemble member:
```
foo.member0.time0.nc
foo.member0.time1.nc
...
foo.member1.time0.nc
foo.member1.time1.nc
...
foo.memberM.timeN.nc
```

This raises many of the same issues as #39.


Discussion / hints:

Fix patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 586dfd65..a59bb41c 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -21,6 +21,7 @@ repos:
     rev: 3.8.3
     hooks:
     - id: flake8
+      exclude: pangeo_forge_recipes/recipes
 
 -   repo: https://github.com/asottile/seed-isort-config
     rev: v2.2.0
@@ -31,7 +32,7 @@ repos:
     rev: 'v0.931'
     hooks:
     -   id: mypy
-        exclude: tests
+        exclude: tests,pangeo_forge_recipes/recipes
 
 -   repo: https://github.com/pycqa/isort
     rev: 5.5.4
diff --git a/docs/pangeo_forge_recipes/api_reference.md b/docs/pangeo_forge_recipes/api_reference.md
index 170fa036..ea2b689b 100644
--- a/docs/pangeo_forge_recipes/api_reference.md
+++ b/docs/pangeo_forge_recipes/api_reference.md
@@ -30,7 +30,11 @@
 ```
 
 ```{eval-rst}
-.. autoclass:: pangeo_forge_recipes.patterns.DimIndex
+.. autoclass:: pangeo_forge_recipes.patterns.DimKey
+    :members:
+```
+```{eval-rst}
+.. autoclass:: pangeo_forge_recipes.patterns.DimVal
     :members:
 ```
 
@@ -63,6 +67,13 @@ These are those functions.
     :members:
 ```
 
+```{eval-rst}
+.. autoclass:: pangeo_forge_recipes.aggregation.XarrayCombineAccumulator
+    :members:
+    :special-members: __add__
+```
+
+
 
 ## PTransforms
 
@@ -76,3 +87,11 @@ The [Beam PTransform Style Guide](https://beam.apache.org/contribute/ptransform-
 .. automodule:: pangeo_forge_recipes.transforms
     :members:
 ```
+
+## Combiners
+
+
+```{eval-rst}
+.. automodule:: pangeo_forge_recipes.combiners
+    :members:
+```
diff --git a/pangeo_forge_recipes/aggregation.py b/pangeo_forge_recipes/aggregation.py
new file mode 100644
index 00000000..14bd0ab7
--- /dev/null
+++ b/pangeo_forge_recipes/aggregation.py
@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, Optional
+
+import numpy as np
+
+
+class DatasetCombineError(Exception):
+    pass
+
+
+def empty_xarray_schema():
+    return {"attrs": None, "coords": None, "data_vars": None, "dims": None, "chunks": None}
+
+
+@dataclass
+class XarrayCombineAccumulator:
+    """An object used to help combine Xarray schemas.
+
+    :param schema: A schema to initialize the accumulator with.
+    :param concat_dim: If set, this accumulator applies concat rules.
+       Otherwise applies merge rules.
+    """
+
+    schema: dict = field(default_factory=empty_xarray_schema)
+    concat_dim: Optional[str] = None
+
+    def add_input(self, s: dict, position: int) -> None:
+        s = s.copy()  # avoid modifying input
+        if "chunks" not in s:
+            s["chunks"] = {}
+        else:
+            s["chunks"] = s["chunks"].copy()
+        if self.concat_dim:
+            assert (
+                self.concat_dim not in s["chunks"]
+            ), "Concat dim should be unchunked for new input"
+            s["chunks"][self.concat_dim] = {position: s["dims"][self.concat_dim]}
+        self.schema = _combine_xarray_schemas(self.schema, s, concat_dim=self.concat_dim)
+
+    def __add__(self, other: XarrayCombineAccumulator) -> XarrayCombineAccumulator:
+        if other.concat_dim != self.concat_dim:
+            raise DatasetCombineError("Can't merge accumulators with different concat_dims")
+        new_schema = _combine_xarray_schemas(self.schema, other.schema, self.concat_dim)
+        return XarrayCombineAccumulator(new_schema, self.concat_dim)
+
+
+def _combine_xarray_schemas(s1: dict, s2: dict, concat_dim: Optional[str] = None):
+    dims = _combine_dims(s1["dims"], s2["dims"], concat_dim)
+    chunks = _combine_chunks(s1["chunks"], s2["chunks"], concat_dim)
+    attrs = _combine_attrs(s1["attrs"], s2["attrs"])
+    data_vars = _combine_vars(s1["data_vars"], s2["data_vars"], concat_dim)
+    coords = _combine_vars(s1["coords"], s2["coords"], concat_dim, allow_both=True)
+    return {
+        "attrs": attrs,
+        "coords": coords,
+        "data_vars": data_vars,
+        "dims": dims,
+        "chunks": chunks,
+    }
+
+
+def _combine_dims(
+    d1: Dict[str, int], d2: Dict[str, int], concat_dim: Optional[str]
+) -> Dict[str, int]:
+    if d1 is None:
+        return d2
+    all_dims = set(d1) | set(d2)
+    new_dims = {}
+    for dim in all_dims:
+        l1 = d1.get(dim, 0)
+        l2 = d2.get(dim, 0)
+        if dim == concat_dim:
+            dim_len = l1 + l2
+        elif l1 != l2:
+            raise DatasetCombineError(f"Dimensions for {dim} have different sizes: {l1}, {l2}")
+        else:
+            dim_len = l1
+        new_dims[dim] = dim_len
+    return new_dims
+
+
+ChunkDict = Dict[str, Dict[int, int]]
+# chunks is a dict like
+# {"lon": {0: 5, 1: 5}}
+#  dim_name: {position, chunk_len}
+
+
+def _combine_chunks(c1: ChunkDict, c2: ChunkDict, concat_dim: Optional[str]) -> ChunkDict:
+    if c1 is None:
+        return c2
+
+    chunks = {}
+    if set(c1) != set(c2):
+        raise DatasetCombineError("Expect the same dims in both chunk sets")
+    for dim in c1:
+        if dim == concat_dim:
+            # merge chunks
+            # check for overlapping keys
+            if set(c1[dim]) & set(c2[dim]):
+                raise DatasetCombineError("Found overlapping keys in concat_dim")
+            chunks[dim] = {**c1[dim], **c2[dim]}
+        else:
+            if c1[dim] != c2[dim]:
+                raise DatasetCombineError("Non concat_dim chunks must be the same")
+            chunks[dim] = c1[dim]
+    return chunks
+
+
+def _combine_attrs(a1: dict, a2: dict) -> dict:
+    if a1 is None:
+        return a2
+    # for now, only keep attrs that are the same in both datasets
+    common_attrs = set(a1) & set(a2)
+    new_attrs = {}
+    for key in common_attrs:
+        if a1[key] == a2[key]:
+            new_attrs[key] = a1[key]
+    return new_attrs
+
+
+def _combine_dtype(d1, d2):
+    return str(np.promote_types(d1, d2))
+
+
+def _combine_vars(v1, v2, concat_dim, allow_both=False):
+    if v1 is None:
+        return v2
+    all_vars = set(v1) | set(v2)
+    new_vars = {}
+    for vname in all_vars:
+        if vname not in v1:
+            new_vars[vname] = v2[vname]
+        elif vname not in v2:
+            new_vars[vname] = v1[vname]
+        else:
+            if concat_dim is None and not allow_both:
+                raise DatasetCombineError(f"Can't merge datasets with the same variable {vname}")
+            attrs = _combine_attrs(v1[vname]["attrs"], v2[vname]["attrs"])
+            dtype = _combine_dtype(v1[vname]["dtype"], v2[vname]["dtype"])
+            (d1, s1), (d2, s2) = (
+                (v1[vname]["dims"], v1[vname]["shape"]),
+                (v2[vname]["dims"], v2[vname]["shape"]),
+            )
+            if d1 != d2:
+                # should we make this logic insensitive to permutations?
+                raise DatasetCombineError(f"Can't merge variables with different dims {d1}, {d2}")
+            dims = d1
+            shape = []
+            for dname, l1, l2 in zip(dims, s1, s2):
+                if dname == concat_dim:
+                    shape.append(l1 + l2)
+                elif l1 != l2:
+                    raise DatasetCombineError(
+                        f"Can't merge variables with different shapes {s1}, {s2}"
+                    )
+                else:
+                    shape.append(l1)
+            new_vars[vname] = {"dims": dims, "attrs": attrs, "dtype": dtype, "shape": tuple(shape)}
+    return new_vars
diff --git a/pangeo_forge_recipes/combiners.py b/pangeo_forge_recipes/combiners.py
new file mode 100644
index 00000000..0d31ff5f
--- /dev/null
+++ b/pangeo_forge_recipes/combiners.py
@@ -0,0 +1,44 @@
+import operator
+from dataclasses import dataclass
+from functools import reduce
+from typing import Sequence, Tuple
+
+import apache_beam as beam
+
+from .aggregation import XarrayCombineAccumulator
+from .patterns import CombineOp, DimKey, Index
+
+
+@dataclass
+class CombineXarraySchemas(beam.CombineFn):
+    """A beam ``CombineFn`` which we can use to combine multiple xarray schemas
+    along a single dimension
+
+    :param dim_key: The dimension along which to combine
+    """
+
+    dim_key: DimKey
+
+    def get_position(self, index: Index) -> int:
+        return index[self.dim_key].position
+
+    def create_accumulator(self) -> XarrayCombineAccumulator:
+        concat_dim = self.dim_key.name if self.dim_key.operation == CombineOp.CONCAT else None
+        return XarrayCombineAccumulator(concat_dim=concat_dim)
+
+    def add_input(self, accumulator: XarrayCombineAccumulator, item: Tuple[Index, dict]):
+        index, schema = item
+        position = self.get_position(index)
+        accumulator.add_input(schema, position)
+        return accumulator
+
+    def merge_accumulators(
+        self, accumulators: Sequence[XarrayCombineAccumulator]
+    ) -> XarrayCombineAccumulator:
+        if len(accumulators) == 1:
+            return accumulators[0]
+        # mypy did not like sum(accumulators)
+        return reduce(operator.add, accumulators)
+
+    def extract_output(self, accumulator) -> dict:
+        return accumulator.schema
diff --git a/pangeo_forge_recipes/patterns.py b/pangeo_forge_recipes/patterns.py
index 19ba1a4b..e7fac4bf 100644
--- a/pangeo_forge_recipes/patterns.py
+++ b/pangeo_forge_recipes/patterns.py
@@ -2,25 +2,13 @@
 Filename / URL patterns.
 """
 import inspect
-from dataclasses import asdict, dataclass, field, replace
+from dataclasses import dataclass, field, replace
 from enum import Enum, auto
 from hashlib import sha256
 from itertools import product
-from typing import (
-    Any,
-    Callable,
-    ClassVar,
-    Dict,
-    FrozenSet,
-    Iterator,
-    List,
-    Optional,
-    Sequence,
-    Tuple,
-    Union,
-)
-
-from .serialization import dataclass_sha256, dict_drop_empty, dict_to_sha256
+from typing import Any, Callable, ClassVar, Dict, Iterator, List, Optional, Sequence, Tuple, Union
+
+from .serialization import dict_drop_empty, dict_to_sha256
 
 
 class CombineOp(Enum):
@@ -69,62 +57,69 @@ class MergeDim:
     operation: ClassVar[CombineOp] = CombineOp.MERGE
 
 
+# would it be simpler to just use a tuple?
 @dataclass(frozen=True, order=True)
-class DimIndex:
-    """Object used to index a single dimension of a FilePattern or Recipe Chunks.
-
-    :param name: The name of the dimension.
-    :param index: The position of the item within the sequence.
-    :param sequence_len: The total length of the sequence.
-    :param operation: What type of Combine Operation does this dimension represent.
-    :param start: The explicit absolute array index start location of the data in this chunk,
-      in respect to the full dataset.
-    :param stop: The explicit absolute array index stop location of the data in this chunk,
-      in respect to the full dataset.
+class DimKey:
+    """
+    :param name: The name of the dimension we are combining over.
+    :param operation: What type of combination this is (merge or concat)
     """
 
     name: str
-    index: int
-    sequence_len: int
     operation: CombineOp
+
+
+@dataclass(frozen=True, order=True)
+class DimVal:
+    """
+    :param position: Where this item lies within the sequence.
+    :param start: Where the starting array index for the item.
+    :param stop: The ending array index for the item.
+    """
+
+    position: int
     start: Optional[int] = None
     stop: Optional[int] = None
 
-    def __str__(self):
-        return f"{self.name}-{self.index}"
 
-    def __post_init__(self):
-        assert self.sequence_len > 0
-        assert self.index >= 0
-        assert self.index < self.sequence_len
-        if self.start:
-            assert self.stop is not None
-            assert self.start < self.stop
+# Alternative way of specifying type
+# Index = dict[DimKey, DimVal]
+
+
+class Index(Dict[DimKey, DimVal]):
+    """An Index is a special sort of dictionary which describes a position within
+    a multidimensional set.
+
+    - The key is a :class:`DimKey` which tells us which dimension we are addressing.
+    - The value is a :class:`DimVal` which tells us where we are within that dimension.
+
+    This object is hashable and deterministically serializable.
+    """
+
+    def __hash__(self):
+        return hash(tuple(self.__getstate__()))
 
+    def __getstate__(self):
+        return sorted(self.items())
 
-class Index(FrozenSet[DimIndex]):
-    pass
+    def __setstate__(self, state):
+        self.__init__({k: v for k, v in state})
 
 
 CombineDim = Union[MergeDim, ConcatDim]
-FilePatternIndex = Index
 
 
-def augment_index_with_start_stop(index: DimIndex, item_lens: List[int]) -> DimIndex:
+def augment_index_with_start_stop(dim_val: DimVal, item_lens: List[int]) -> DimVal:
     """Take an index _without_ start / stop and add them based on the lens defined in sequence_lens.
 
     :param index: The ``DimIndex`` instance to augment.
     :param item_lens: A list of integer lengths for all items in the sequence.
     """
 
-    assert len(item_lens) == index.sequence_len
-    start = sum(item_lens[: index.index])
-    stop = start + item_lens[index.index]
+    start = sum(item_lens[: dim_val.position])
+    stop = start + item_lens[dim_val.position]
 
-    kw = asdict(index)
-    kw["start"] = start
-    kw["stop"] = stop
-    return DimIndex(**kw)
+    return DimVal(dim_val.position, start, stop)
 
 
 class AutoName(Enum):
@@ -228,31 +223,28 @@ def concat_sequence_lens(self) -> Dict[str, Optional[int]]:
             for dim_name, nitems in self.nitems_per_input.items()
         }
 
-    def __getitem__(self, indexer: FilePatternIndex) -> str:
+    def __getitem__(self, indexer: Index) -> str:
         """Get a filename path for a particular key."""
         assert len(indexer) == len(self.combine_dims)
         format_function_kwargs = {}
-        for idx in indexer:
+        for dimkey, dimval in indexer.items():
             dims = [
                 dim
                 for dim in self.combine_dims
-                if dim.name == idx.name and dim.operation == idx.operation
+                if dim.name == dimkey.name and dim.operation == dimkey.operation
             ]
             if len(dims) != 1:
-                raise KeyError(r"Could not valid combine_dim for indexer {idx}")
+                raise KeyError(r"Could not valid combine_dim for indexer {idx_key}")
             dim = dims[0]
-            format_function_kwargs[dim.name] = dim.keys[idx.index]
+            format_function_kwargs[dim.name] = dim.keys[dimval.position]
         fname = self.format_function(**format_function_kwargs)
         return fname
 
-    def __iter__(self) -> Iterator[FilePatternIndex]:
+    def __iter__(self) -> Iterator[Index]:
         """Iterate over all keys in the pattern."""
         for val in product(*[range(n) for n in self.shape]):
             index = Index(
-                (
-                    DimIndex(op.name, v, len(op.keys), op.operation)
-                    for op, v in zip(self.combine_dims, val)
-                )
+                {DimKey(op.name, op.operation): DimVal(v) for op, v in zip(self.combine_dims, val)}
             )
             yield index
 
@@ -339,11 +331,9 @@ def pattern_blockchain(pattern: FilePattern) -> List[bytes]:
 
     blockchain = [root_sha256]
     for k, v in pattern.items():
-        key_hash = b"".join(
-            sorted([dataclass_sha256(dimindex, ignore_keys=["sequence_len"]) for dimindex in k])
-        )
+        # key is no longer part of the hash
         value_hash = sha256(v.encode("utf-8")).digest()
-        new_hash = key_hash + value_hash
+        new_hash = value_hash
         new_block = sha256(blockchain[-1] + new_hash).digest()
         blockchain.append(new_block)
 
diff --git a/pangeo_forge_recipes/recipes/xarray_zarr.py b/pangeo_forge_recipes/recipes/xarray_zarr.py
deleted file mode 100644
index bb00bf9c..00000000
--- a/pangeo_forge_recipes/recipes/xarray_zarr.py
+++ /dev/null
@@ -1,883 +0,0 @@
-"""
-A Pangeo Forge Recipe
-"""
-from __future__ import annotations
-
-import itertools
-import logging
-import os
-import warnings
-from contextlib import ExitStack, contextmanager
-from dataclasses import dataclass, field, replace
-from itertools import chain, product
-from math import ceil
-from typing import Callable, Dict, Iterator, List, Optional, Sequence, Set, Tuple
-
-import dask
-import fsspec
-import numpy as np
-import xarray as xr
-import zarr
-
-from ..chunk_grid import ChunkGrid
-from ..executors.base import Pipeline, Stage
-from ..patterns import CombineOp, DimIndex, FilePattern, FileType, Index
-from ..reference import create_hdf5_reference, unstrip_protocol
-from ..storage import FSSpecTarget, MetadataTarget, file_opener
-from ..utils import calc_subsets, fix_scalar_attr_encoding, lock_for_conflicts
-from .base import BaseRecipe, FilePatternMixin, StorageMixin
-
-# use this filename to store global recipe metadata in the metadata_cache
-# it will be written once (by prepare_target) and read many times (by store_chunk)
-_GLOBAL_METADATA_KEY = "pangeo-forge-recipe-metadata.json"
-_ARRAY_DIMENSIONS = "_ARRAY_DIMENSIONS"
-MAX_MEMORY = (
-    int(os.getenv("PANGEO_FORGE_MAX_MEMORY"))  # type: ignore
-    if os.getenv("PANGEO_FORGE_MAX_MEMORY")
-    else 500_000_000
-)
-OPENER_MAP = {
-    FileType.netcdf3: dict(engine="scipy"),
-    FileType.netcdf4: dict(engine="h5netcdf"),
-}
-
-logger = logging.getLogger(__name__)
-
-# Some types that help us keep things organized
-InputKey = Index  # input keys are the same as the file pattern keys
-ChunkKey = Index
-
-SubsetSpec = Dict[str, int]
-# SubsetSpec is a dictionary mapping dimension names to the number of subsets along that dimension
-# (e.g. {'time': 5, 'depth': 2})
-
-
-def _input_metadata_fname(input_key: InputKey) -> str:
-    key_str = "-".join([f"{k.name}_{k.index}" for k in sorted(input_key)])
-    return "input-meta-" + key_str + ".json"
-
-
-def _input_reference_fname(input_key: InputKey) -> str:
-    key_str = "-".join([f"{k.name}_{k.index}" for k in sorted(input_key)])
-    return "input-reference-" + key_str + ".json"
-
-
-def inputs_for_chunk(
-    chunk_key: ChunkKey, inputs_per_chunk: int, ninputs: int
-) -> Sequence[InputKey]:
-    """For a chunk key, figure out which inputs belong to it.
-    Returns at least one InputKey."""
-
-    merge_dims = [dim_idx for dim_idx in chunk_key if dim_idx.operation == CombineOp.MERGE]
-    concat_dims = [dim_idx for dim_idx in chunk_key if dim_idx.operation == CombineOp.CONCAT]
-    # Ignore subset dims, we don't need them here
-    # subset_dims = [dim_idx for dim_idx in chunk_key if dim_idx.operation == CombineOp.SUBSET]
-    assert len(merge_dims) <= 1
-    if len(merge_dims) == 1:
-        merge_dim = merge_dims[0]  # type: Optional[DimIndex]
-    else:
-        merge_dim = None
-    assert len(concat_dims) == 1
-    concat_dim = concat_dims[0]
-
-    input_keys = []
-
-    for n in range(inputs_per_chunk):
-        input_index = (inputs_per_chunk * concat_dim.index) + n
-        if input_index >= ninputs:
-            break
-        input_concat_dim = DimIndex(concat_dim.name, input_index, ninputs, CombineOp.CONCAT)
-        input_key = [input_concat_dim]
-        if merge_dim is not None:
-            input_key.append(merge_dim)
-        input_keys.append(Index(input_key))
-
-    return input_keys
-
-
-def expand_target_dim(target: FSSpecTarget, concat_dim: Optional[str], dimsize: int) -> None:
-    target_mapper = target.get_mapper()
-    zgroup = zarr.open_group(target_mapper)
-    ds = open_target(target)
-    sequence_axes = {
-        v: ds[v].get_axis_num(concat_dim) for v in ds.variables if concat_dim in ds[v].dims
-    }
-
-    for v, axis in sequence_axes.items():
-        arr = zgroup[v]
-        shape = list(arr.shape)
-        shape[axis] = dimsize
-        logger.debug(f"resizing array {v} to shape {shape}")
-        arr.resize(shape)
-
-
-def open_target(target: FSSpecTarget) -> xr.Dataset:
-    return xr.open_zarr(target.get_mapper())
-
-
-def input_position(input_key: InputKey) -> int:
-    """Return the position of the input within the input sequence."""
-    for dim_idx in input_key:
-        # assumes there is one and only one concat dim
-        if dim_idx.operation == CombineOp.CONCAT:
-            return dim_idx.index
-    return -1  # make mypy happy
-
-
-def chunk_position(chunk_key: ChunkKey) -> int:
-    """Return the position of the input within the input sequence."""
-    concat_idx = -1
-    for dim_idx in chunk_key:
-        # assumes there is one and only one concat dim
-        if dim_idx.operation == CombineOp.CONCAT:
-            concat_idx = dim_idx.index
-            concat_dim = dim_idx.name
-    if concat_idx == -1:
-        raise ValueError("Couldn't find concat_dim")
-    subset_idx = 0
-    subset_factor = 1
-    for dim_idx in chunk_key:
-        if dim_idx.operation == CombineOp.SUBSET:
-            if dim_idx.name == concat_dim:
-                subset_idx = dim_idx.index
-                subset_factor = dim_idx.sequence_len
-    return subset_factor * concat_idx + subset_idx
-
-
-def cache_input(input_key: InputKey, *, config: XarrayZarrRecipe) -> None:
-    if config.cache_inputs:
-        if config.file_pattern.file_type == FileType.opendap:
-            raise ValueError("Can't cache opendap inputs")
-        if config.storage_config.cache is None:
-            raise ValueError("input_cache is not set.")
-        logger.info(f"Caching input '{input_key!s}'")
-        fname = config.file_pattern[input_key]
-        config.storage_config.cache.cache_file(
-            fname,
-            config.file_pattern.query_string_secrets,
-            **config.file_pattern.fsspec_open_kwargs,
-        )
-
-    if config.cache_metadata:
-        if config.storage_config.metadata is None:
-            raise ValueError("metadata_cache is not set.")
-
-        if not _input_metadata_fname(input_key) in config.storage_config.metadata:
-            with open_input(input_key, config=config) as ds:
-                logger.info(f"Caching metadata for input '{input_key!s}'")
-                input_metadata = ds.to_dict(data=False)
-                config.storage_config.metadata[_input_metadata_fname(input_key)] = input_metadata
-        else:
-            logger.info(f"Metadata already cached for input '{input_key!s}'")
-
-    if config.open_input_with_kerchunk:
-        if config.file_pattern.file_type == FileType.opendap:
-            raise ValueError("Can't make references for opendap inputs")
-        if config.storage_config.metadata is None:
-            raise ValueError("Can't make references; no metadata cache assigned")
-        fname = config.file_pattern[input_key]
-
-        ref_fname = _input_reference_fname(input_key)
-
-        if ref_fname in config.storage_config.metadata:
-            logger.info("Metadata is already cached with kerchunk.")
-            return
-
-        if config.storage_config.cache is None:
-            protocol = fsspec.utils.get_protocol(fname)
-            url = unstrip_protocol(fname, protocol)
-        else:
-            url = unstrip_protocol(
-                config.storage_config.cache._full_path(fname),
-                config.storage_config.cache.fs.protocol,
-            )
-        with file_opener(
-            fname,
-            cache=config.storage_config.cache,
-            copy_to_local=config.copy_input_to_local_file,
-            bypass_open=False,
-            secrets=config.file_pattern.query_string_secrets,
-            **config.file_pattern.fsspec_open_kwargs,
-        ) as fp:
-            ref_data = create_hdf5_reference(fp, url, fname)
-        config.storage_config.metadata[ref_fname] = ref_data
-
-
-def region_and_conflicts_for_chunk(
-    config: XarrayZarrRecipe, chunk_key: ChunkKey
-) -> Tuple[Dict[str, slice], Dict[str, Set[int]]]:
-    # return a dict suitable to pass to xr.to_zarr(region=...)
-    # specifies where in the overall array to put this chunk's data
-    # also return the conflicts with other chunks
-
-    if config.nitems_per_input:
-        input_sequence_lens = (config.nitems_per_input,) * config.file_pattern.dims[
-            config.concat_dim
-        ]  # type: ignore
-    else:
-        assert config.storage_config.metadata is not None  # for mypy
-        global_metadata = config.storage_config.metadata[_GLOBAL_METADATA_KEY]
-        input_sequence_lens = global_metadata["input_sequence_lens"]
-    total_len = sum(input_sequence_lens)
-
-    # for now this will just have one key since we only allow one concat_dim
-    # but it could expand to accomodate multiple concat dims
-    chunk_index = {config.concat_dim: chunk_position(chunk_key)}
-
-    input_chunk_grid = ChunkGrid({config.concat_dim: input_sequence_lens})
-    if config.subset_inputs and config.concat_dim in config.subset_inputs:
-        assert (
-            config.inputs_per_chunk == 1
-        ), "Doesn't make sense to have multiple inputs per chunk plus subsetting"
-        chunk_grid = input_chunk_grid.subset(config.subset_inputs)
-    elif config.inputs_per_chunk > 1:
-        chunk_grid = input_chunk_grid.consolidate({config.concat_dim: config.inputs_per_chunk})
-    else:
-        chunk_grid = input_chunk_grid
-    assert chunk_grid.shape[config.concat_dim] == total_len
-
-    region = chunk_grid.chunk_index_to_array_slice(chunk_index)
-
-    assert config.concat_dim_chunks is not None
-    target_grid = ChunkGrid.from_uniform_grid(
-        {config.concat_dim: (config.concat_dim_chunks, total_len)}
-    )
-    conflicts = chunk_grid.chunk_conflicts(chunk_index, target_grid)
-
-    return region, conflicts
-
-
-@contextmanager
-def open_input(input_key: InputKey, *, config: XarrayZarrRecipe) -> xr.Dataset:
-    fname = config.file_pattern[input_key]
-    logger.info(f"Opening input with Xarray {input_key!s}: '{fname}'")
-
-    if config.file_pattern.file_type == FileType.opendap:
-        if config.cache_inputs:
-            raise ValueError("Can't cache opendap inputs")
-        if config.copy_input_to_local_file:
-            raise ValueError("Can't copy opendap inputs to local file")
-        if config.open_input_with_kerchunk:
-            raise ValueError("Can't open opendap inputs with kerchunk")
-
-    if config.open_input_with_kerchunk:
-        if config.storage_config.metadata is None:
-            raise ValueError("metadata_cache is not set.")
-        from fsspec.implementations.reference import ReferenceFileSystem
-
-        reference_data = config.storage_config.metadata[_input_reference_fname(input_key)]
-        # TODO: figure out how to set this for the cache target
-        remote_protocol = fsspec.utils.get_protocol(next(config.file_pattern.items())[1])
-        ref_fs = ReferenceFileSystem(
-            reference_data, remote_protocol=remote_protocol, skip_instance_cache=True
-        )
-        mapper = ref_fs.get_mapper("/")
-        # Doesn't really need to be a context manager, but that's how this function works
-        with dask.config.set(scheduler="single-threaded"):  # make sure we don't use a scheduler
-            with xr.open_dataset(mapper, engine="zarr", chunks={}, consolidated=False) as ds:
-                logger.debug("successfully opened reference dataset with zarr")
-
-                if config.delete_input_encoding:
-                    for var in ds.variables:
-                        ds[var].encoding = {}
-
-                if config.process_input is not None:
-                    ds = config.process_input(ds, str(fname))
-
-                logger.debug(f"{ds}")
-                yield ds
-
-    else:
-
-        cache = config.storage_config.cache if config.cache_inputs else None
-        bypass_open = True if config.file_pattern.file_type == FileType.opendap else False
-
-        with file_opener(
-            fname,
-            cache=cache,
-            copy_to_local=config.copy_input_to_local_file,
-            bypass_open=bypass_open,
-            secrets=config.file_pattern.query_string_secrets,
-            **config.file_pattern.fsspec_open_kwargs,
-        ) as f:
-            with dask.config.set(scheduler="single-threaded"):  # make sure we don't use a scheduler
-                kw = config.xarray_open_kwargs.copy()
-                file_type = config.file_pattern.file_type
-                if file_type in OPENER_MAP:
-                    if "engine" in kw:
-                        engine_message_base = (
-                            "pangeo-forge-recipes will automatically set the xarray backend for "
-                            f"files of type '{file_type.value}' to '{OPENER_MAP[file_type]}', "
-                        )
-                        warn_matching_msg = engine_message_base + (
-                            "which is the same value you have passed via `xarray_open_kwargs`. "
-                            f"If this input file is actually of type '{file_type.value}', you can "
-                            f"remove `{{'engine': '{kw['engine']}'}}` from `xarray_open_kwargs`. "
-                        )
-                        error_mismatched_msg = engine_message_base + (
-                            f"which is different from the value you have passed via "
-                            "`xarray_open_kwargs`. If this input file is actually of type "
-                            f"'{file_type.value}', please remove `{{'engine': '{kw['engine']}'}}` "
-                            "from `xarray_open_kwargs`. "
-                        )
-                        engine_message_tail = (
-                            f"If this input file is not of type '{file_type.value}', please update"
-                            " this recipe by passing a different value to `FilePattern.file_type`."
-                        )
-                        warn_matching_msg += engine_message_tail
-                        error_mismatched_msg += engine_message_tail
-
-                        if kw["engine"] == OPENER_MAP[file_type]["engine"]:
-                            warnings.warn(warn_matching_msg)
-                        elif kw["engine"] != OPENER_MAP[file_type]["engine"]:
-                            raise ValueError(error_mismatched_msg)
-                    else:
-                        kw.update(OPENER_MAP[file_type])
-                logger.debug(f"about to enter xr.open_dataset context on {f}")
-                try:
-                    with xr.open_dataset(f, **kw) as ds:
-                        logger.debug("successfully opened dataset")
-                        ds = fix_scalar_attr_encoding(ds)
-
-                        if config.delete_input_encoding:
-                            for var in ds.variables:
-                                ds[var].encoding = {}
-
-                        if config.process_input is not None:
-                            ds = config.process_input(ds, str(fname))
-
-                        logger.debug(f"{ds}")
-                        yield ds
-                except OSError as e:
-                    if "unable to open file (file signature not found)" in str(e).lower():
-                        oserror_message_base = (
-                            f"Unable to open file {f.path} "  # type: ignore
-                            f"with `{{engine: {kw.get('engine')}}}`, "
-                        )
-                        if "engine" in config.xarray_open_kwargs:
-                            oserror_message = oserror_message_base + (
-                                "which was set explicitly via `xarray_open_kwargs`. Please remove "
-                                f"`{{engine: {kw.get('engine')}}}` from `xarray_open_kwargs`."
-                            )
-                        elif file_type == FileType.netcdf4:
-                            oserror_message = oserror_message_base + (
-                                "which was set automatically based on the fact that "
-                                "`FilePattern.file_type` is using the default value of 'netcdf4'. "
-                                "It seems likely that this input file is in NetCDF3 format. If "
-                                "that is the case, please re-instantiate your `FilePattern` with "
-                                '`FilePattern(..., file_type="netcdf3")`.'
-                            )
-                        raise OSError(oserror_message) from e
-                    else:
-                        raise e
-
-
-def subset_dataset(ds: xr.Dataset, subset_spec: DimIndex) -> xr.Dataset:
-    assert subset_spec.operation == CombineOp.SUBSET
-    dim = subset_spec.name
-    dim_len = ds.dims[dim]
-    subset_lens = calc_subsets(dim_len, subset_spec.sequence_len)
-    start = sum(subset_lens[: subset_spec.index])
-    stop = sum(subset_lens[: (subset_spec.index + 1)])
-    subset_slice = slice(start, stop)
-    indexer = {dim: subset_slice}
-    logger.debug(f"Subsetting dataset with indexer {indexer}")
-    return ds.isel(**indexer)
-
-
-@contextmanager
-def open_chunk(chunk_key: ChunkKey, *, config: XarrayZarrRecipe) -> xr.Dataset:
-    logger.info(f"Opening inputs for chunk {chunk_key!s}")
-    ninputs = config.file_pattern.dims[config.file_pattern.concat_dims[0]]
-    inputs = inputs_for_chunk(chunk_key, config.inputs_per_chunk, ninputs)
-
-    # need to open an unknown number of contexts at the same time
-    with ExitStack() as stack:
-        dsets = [stack.enter_context(open_input(input_key, config=config)) for input_key in inputs]
-
-        # subset before chunking; hopefully lazy
-        subset_dims = [dim_idx for dim_idx in chunk_key if dim_idx.operation == CombineOp.SUBSET]
-
-        for subset_dim in subset_dims:
-            logger.info(f"Subsetting input according to {subset_dim}")
-            dsets = [subset_dataset(ds, subset_dim) for ds in dsets]
-
-        # explicitly chunking prevents eager evaluation during concat
-        dsets = [ds.chunk() for ds in dsets]
-
-        logger.info(f"Combining inputs for chunk '{chunk_key!s}'")
-        if len(dsets) > 1:
-            # During concat, attributes and encoding are taken from the first dataset
-            # https://github.com/pydata/xarray/issues/1614
-            with dask.config.set(scheduler="single-threaded"):  # make sure we don't use a scheduler
-                ds = xr.concat(dsets, config.concat_dim, **config.xarray_concat_kwargs)
-        elif len(dsets) == 1:
-            ds = dsets[0]
-        else:  # pragma: no cover
-            assert False, "Should never happen"
-
-        if config.process_chunk is not None:
-            with dask.config.set(scheduler="single-threaded"):  # make sure we don't use a scheduler
-                ds = config.process_chunk(ds)
-
-        with dask.config.set(scheduler="single-threaded"):  # make sure we don't use a scheduler
-            logger.debug(f"{ds}")
-
-        if config.target_chunks:
-            # The input may be too large to process in memory at once, so
-            # rechunk it to the target chunks.
-            ds = ds.chunk(config.target_chunks)
-        yield ds
-
-
-def get_input_meta(
-    metadata_cache: Optional[MetadataTarget],
-    file_pattern: FilePattern,
-) -> Dict:
-    # getitems should be async; much faster than serial calls
-    if metadata_cache is None:
-        raise ValueError("metadata_cache is not set.")
-    return metadata_cache.getitems([_input_metadata_fname(k) for k in file_pattern])
-
-
-def _get_fname_from_error_pos(pos: Tuple[int, int], file_pattern: FilePattern) -> str:
-    kwargs = {}
-    for idx, p in enumerate(pos):
-        dim = file_pattern.combine_dims[idx]
-        kwargs.update({dim.name: dim.keys[p]})
-
-    return file_pattern.format_function(**kwargs)
-
-
-def calculate_sequence_lens(
-    nitems_per_input: Optional[int],
-    file_pattern: FilePattern,
-    metadata_cache: Optional[MetadataTarget],
-) -> List[int]:
-
-    assert len(file_pattern.concat_dims) == 1
-    concat_dim = file_pattern.concat_dims[0]
-
-    if nitems_per_input:
-        concat_dim = file_pattern.concat_dims[0]
-        return list((nitems_per_input,) * file_pattern.dims[concat_dim])
-
-    # read per-input metadata; this is distinct from global metadata
-    # get the sequence length of every file
-    input_meta = get_input_meta(metadata_cache, file_pattern)
-    # use a numpy array to allow reshaping
-    all_lens = np.array([m["dims"][concat_dim] for m in input_meta.values()])
-    all_lens.shape = list(file_pattern.dims.values())
-
-    if len(all_lens.shape) == 1:
-        all_lens = np.expand_dims(all_lens, 1)
-
-    # check that all lens are the same along the concat dim
-    unique_vec, unique_idx, unique_counts = np.unique(
-        all_lens, axis=1, return_index=True, return_counts=True
-    )
-
-    if len(unique_idx) > 1:
-        correct_col = all_lens[:, unique_idx[np.argmax(unique_counts)]]
-        err_idx = np.where(np.expand_dims(correct_col, 1) != all_lens)
-        err_pos = list(zip(*err_idx))
-        # present a list of problematic files to the user, given the file pattern.
-        file_list = "\n".join(
-            [f"- {_get_fname_from_error_pos(epos, file_pattern)!r}" for epos in err_pos]
-        )
-        raise ValueError(
-            f"Inconsistent sequence lengths between indices {unique_idx} of the concat dim."
-            f"\nValue(s) {all_lens[err_idx]} at position(s) {err_pos} are different from the rest."
-            f"\nPlease check the following file(s) for data errors:"
-            f"\n{file_list}"
-            f"\n{all_lens}"
-        )
-    return np.atleast_1d(unique_vec.squeeze()).tolist()
-
-
-def prepare_target(*, config: XarrayZarrRecipe) -> None:
-    if config.storage_config.target is None:
-        raise ValueError("Cannot proceed without a target")
-    try:
-        ds = open_target(config.storage_config.target)
-        logger.info("Found an existing dataset in target")
-        logger.debug(f"{ds}")
-
-        if config.target_chunks:
-            # TODO: check that target_chunks id compatibile with the
-            # existing chunks
-            pass
-    except (FileNotFoundError, IOError, zarr.errors.GroupNotFoundError):
-        logger.info("Creating a new dataset in target")
-
-        # need to rewrite this as an append loop
-
-        def filter_init_chunks(chunk_key):
-            for dim_idx in chunk_key:
-                if (dim_idx.operation != CombineOp.MERGE) and (dim_idx.index > 0):
-                    return False
-            return True
-
-        init_chunks = list(filter(filter_init_chunks, config.iter_chunks()))
-
-        for chunk_key in init_chunks:
-            with open_chunk(chunk_key, config=config) as ds:
-                # ds is already chunked
-
-                # https://github.com/pydata/xarray/blob/5287c7b2546fc8848f539bb5ee66bb8d91d8496f/xarray/core/variable.py#L1069
-                for v in ds.variables:
-                    if config.target_chunks:
-                        this_var = ds[v]
-                        chunks = {
-                            this_var.get_axis_num(dim): chunk
-                            for dim, chunk in config.target_chunks.items()
-                            if dim in this_var.dims
-                        }
-                        encoding_chunks = tuple(
-                            chunks.get(n, s) for n, s in enumerate(this_var.shape)
-                        )
-                    else:
-                        encoding_chunks = ds[v].shape
-                    logger.debug(f"Setting variable {v} encoding chunks to {encoding_chunks}")
-                    ds[v].encoding["chunks"] = encoding_chunks
-
-                # load all variables that don't have the sequence dim in them
-                # these are usually coordinates.
-                # Variables that are loaded will be written even with compute=False
-                # TODO: make this behavior customizable
-                for v in ds.variables:
-                    if config.concat_dim not in ds[v].dims:
-                        ds[v].load()
-
-                target_mapper = config.storage_config.target.get_mapper()
-                logger.info(f"Storing dataset in {config.storage_config.target.root_path}")
-                logger.debug(f"{ds}")
-                with warnings.catch_warnings():
-                    warnings.simplefilter(
-                        "ignore"
-                    )  # suppress the warning that comes with safe_chunks
-                    ds.to_zarr(target_mapper, mode="a", compute=False, safe_chunks=False)
-
-    # Regardless of whether there is an existing dataset or we are creating a new one,
-    # we need to expand the concat_dim to hold the entire expected size of the data
-    input_sequence_lens = calculate_sequence_lens(
-        config.nitems_per_input,
-        config.file_pattern,
-        config.storage_config.metadata,
-    )
-    n_sequence = sum(input_sequence_lens)
-    logger.info(f"Expanding target concat dim '{config.concat_dim}' to size {n_sequence}")
-    expand_target_dim(config.storage_config.target, config.concat_dim, n_sequence)
-    # TODO: handle possible subsetting
-    # The init chunks might not cover the whole dataset along multiple dimensions!
-
-    if config.cache_metadata:
-        # if nitems_per_input is not constant, we need to cache this info
-        assert config.storage_config.metadata is not None  # for mypy
-        recipe_meta = {"input_sequence_lens": input_sequence_lens}
-        config.storage_config.metadata[_GLOBAL_METADATA_KEY] = recipe_meta
-
-
-def store_chunk(chunk_key: ChunkKey, *, config: XarrayZarrRecipe) -> None:
-    if config.storage_config.target is None:
-        raise ValueError("target has not been set.")
-
-    with open_chunk(chunk_key, config=config) as ds_chunk:
-        # writing a region means that all the variables MUST have concat_dim
-        to_drop = [v for v in ds_chunk.variables if config.concat_dim not in ds_chunk[v].dims]
-        ds_chunk = ds_chunk.drop_vars(to_drop)
-
-        target_mapper = config.storage_config.target.get_mapper()
-        write_region, conflicts = region_and_conflicts_for_chunk(config, chunk_key)
-
-        zgroup = zarr.open_group(target_mapper)
-        for vname, var_coded in ds_chunk.variables.items():
-            zarr_array = zgroup[vname]
-            # get encoding for variable from zarr attributes
-            # could this backfire some way?
-            var_coded.encoding.update(zarr_array.attrs)
-            # just delete all attributes from the var;
-            # they are not used anyway, and there can be conflicts
-            # related to xarray.coding.variables.safe_setitem
-            var_coded.attrs = {}
-            with dask.config.set(scheduler="single-threaded"):  # make sure we don't use a scheduler
-                var = xr.backends.zarr.encode_zarr_variable(var_coded)
-                logger.debug(
-                    f"Converting variable {vname} of {var.data.nbytes} bytes to `numpy.ndarray`"
-                )
-                if var.data.nbytes > MAX_MEMORY:
-                    factor = round((var.data.nbytes / MAX_MEMORY), 2)
-                    cdim = config.concat_dim
-                    logger.warning(
-                        f"Variable {vname} of {var.data.nbytes} bytes is {factor} times larger "
-                        f"than specified maximum variable array size of {MAX_MEMORY} bytes. "
-                        f'Consider re-instantiating recipe with `subset_inputs = {{"{cdim}": '
-                        f'{ceil(factor)}}}`. If `len(ds["{cdim}"])` < {ceil(factor)}, '
-                        f'substitute "{cdim}" for any name in ds["{vname}"].dims with length '
-                        f">= {ceil(factor)} or consider subsetting along multiple dimensions."
-                        " Setting PANGEO_FORGE_MAX_MEMORY env variable changes the variable array"
-                        " size which will trigger this warning."
-                    )
-                data = np.asarray(
-                    var.data
-                )  # TODO: can we buffer large data rather than loading it all?
-            zarr_region = tuple(write_region.get(dim, slice(None)) for dim in var.dims)
-            lock_keys = [
-                f"{vname}-{dim}-{c}"
-                for dim, dim_conflicts in conflicts.items()
-                for c in dim_conflicts
-            ]
-            logger.debug(f"Acquiring locks {lock_keys}")
-            with lock_for_conflicts(lock_keys, timeout=config.lock_timeout):
-                logger.info(
-                    f"Storing variable {vname} chunk {chunk_key!s} " f"to Zarr region {zarr_region}"
-                )
-                zarr_array[zarr_region] = data
-
-
-def _gather_coordinate_dimensions(group: zarr.Group) -> List[str]:
-    return list(
-        set(itertools.chain(*(group[var].attrs.get(_ARRAY_DIMENSIONS, []) for var in group)))
-    )
-
-
-def finalize_target(*, config: XarrayZarrRecipe) -> None:
-    if config.storage_config.target is None:
-        raise ValueError("target has not been set.")
-
-    if config.consolidate_dimension_coordinates:
-        logger.info("Consolidating dimension coordinate arrays")
-        target_mapper = config.storage_config.target.get_mapper()
-        group = zarr.open(target_mapper, mode="a")
-        # https://github.com/pangeo-forge/pangeo-forge-recipes/issues/214
-        # filter out the dims from the array metadata not in the Zarr group
-        # to handle coordinateless dimensions.
-        dims = (dim for dim in _gather_coordinate_dimensions(group) if dim in group)
-        for dim in dims:
-            arr = group[dim]
-            attrs = dict(arr.attrs)
-            new = group.array(
-                dim,
-                arr[:],
-                chunks=arr.shape,
-                dtype=arr.dtype,
-                compressor=arr.compressor,
-                fill_value=arr.fill_value,
-                order=arr.order,
-                filters=arr.filters,
-                overwrite=True,
-            )
-            new.attrs.update(attrs)
-
-    if config.consolidate_zarr:
-        logger.info("Consolidating Zarr metadata")
-        target_mapper = config.storage_config.target.get_mapper()
-        zarr.consolidate_metadata(target_mapper)
-
-
-def xarray_zarr_recipe_compiler(recipe: XarrayZarrRecipe) -> Pipeline:
-    stages = [
-        Stage(name="cache_input", function=cache_input, mappable=list(recipe.iter_inputs())),
-        Stage(name="prepare_target", function=prepare_target),
-        Stage(name="store_chunk", function=store_chunk, mappable=list(recipe.iter_chunks())),
-        Stage(name="finalize_target", function=finalize_target),
-    ]
-    return Pipeline(stages=stages, config=recipe)
-
-
-# Notes about dataclasses:
-# - https://www.python.org/dev/peps/pep-0557/#inheritance
-# - https://stackoverflow.com/questions/51575931/class-inheritance-in-python-3-7-dataclasses
-
-
-_deprecation_message = (
-    "This method will be deprecated in v0.8.0. "
-    "Please call the equivalent function directly from the xarray_zarr module."
-)
-
-
-@dataclass
-class XarrayZarrRecipe(BaseRecipe, StorageMixin, FilePatternMixin):
-    """This configuration represents a dataset composed of many individual NetCDF files.
-    This class uses Xarray to read and write data and writes its output to Zarr.
-    The organization of the source files is described by the ``file_pattern``.
-    Currently this recipe supports at most one ``MergeDim`` and one ``ConcatDim``
-    in the File Pattern.
-
-    :param file_pattern: An object which describes the organization of the input files.
-    :param inputs_per_chunk: The number of inputs to use in each chunk along the concat dim.
-       Must be an integer >= 1.
-    :param target_chunks: Desired chunk structure for the targret dataset. This is a dictionary
-       mapping dimension names to chunk size. When using a :class:`patterns.FilePattern` with
-       a :class:`patterns.ConcatDim` that specifies ``n_items_per_file``, then you don't need
-       to include the concat dim in ``target_chunks``.
-    :param storage_config: Defines locations for writing the output dataset, caching temporary data,
-      and for caching metadata for inputs and chunks. All three locations default to
-      ``tempdir.TemporaryDirectory``; this default config can be used for testing and debugging the
-      recipe. In an actual execution context, the default config is re-assigned to point to the
-      destination(s) of choice, which can be any combination of ``fsspec``-compatible storage
-      backends.
-    :param cache_inputs: If ``True``, inputs are copied to ``input_cache`` before
-      opening. If ``False``, try to open inputs directly from their source location.
-    :param copy_input_to_local_file: Whether to copy the inputs to a temporary
-      local file. In this case, a path (rather than file object) is passed to
-      ``xr.open_dataset``. This is required for engines that can't open
-      file-like objects (e.g. pynio).
-    :param consolidate_zarr: Whether to consolidate the resulting Zarr dataset.
-    :param consolidate_dimension_coordinates: Whether to rewrite coordinate variables as a
-        single chunk. We recommend consolidating coordinate variables to avoid
-        many small read requests to get the coordinates in xarray.
-    :param xarray_open_kwargs: Extra options for opening the inputs with Xarray.
-    :param xarray_concat_kwargs: Extra options to pass to Xarray when concatenating
-      the inputs to form a chunk.
-    :param delete_input_encoding: Whether to remove Xarray encoding from variables
-      in the input dataset
-    :param process_input: Function to call on each opened input, with signature
-      `(ds: xr.Dataset, filename: str) -> ds: xr.Dataset`.
-    :param process_chunk: Function to call on each concatenated chunk, with signature
-      `(ds: xr.Dataset) -> ds: xr.Dataset`.
-    :param lock_timeout: The default timeout for acquiring a chunk lock.
-    :param subset_inputs: If set, break each input file up into multiple chunks
-      along dimension according to the specified mapping. For example,
-      ``{'time': 5}`` would split each input file into 5 chunks along the
-      time dimension. Multiple dimensions are allowed.
-    :param open_input_with_kerchunk: If True, use kerchunk
-      to generate a reference filesystem for each input, to be used when opening
-      the file with Xarray as a virtual Zarr dataset.
-    """
-
-    _compiler = xarray_zarr_recipe_compiler
-
-    inputs_per_chunk: int = 1
-    target_chunks: Dict[str, int] = field(default_factory=dict)
-    cache_inputs: Optional[bool] = None
-    copy_input_to_local_file: bool = False
-    consolidate_zarr: bool = True
-    consolidate_dimension_coordinates: bool = True
-    xarray_open_kwargs: dict = field(default_factory=dict)
-    xarray_concat_kwargs: dict = field(default_factory=dict)
-    delete_input_encoding: bool = True
-    process_input: Optional[Callable[[xr.Dataset, str], xr.Dataset]] = None
-    process_chunk: Optional[Callable[[xr.Dataset], xr.Dataset]] = None
-    lock_timeout: Optional[int] = None
-    subset_inputs: SubsetSpec = field(default_factory=dict)
-    open_input_with_kerchunk: bool = False
-
-    # internal attributes not meant to be seen or accessed by user
-    concat_dim: str = field(default_factory=str, repr=False, init=False)
-    """The concatenation dimension name."""
-
-    concat_dim_chunks: Optional[int] = field(default=None, repr=False, init=False)
-    """The desired chunking along the sequence dimension."""
-
-    init_chunks: List[ChunkKey] = field(default_factory=list, repr=False, init=False)
-    """List of chunks needed to initialize the recipe."""
-
-    cache_metadata: bool = field(default=False, repr=False, init=False)
-    """Whether metadata caching is needed."""
-
-    nitems_per_input: Optional[int] = field(default=None, repr=False, init=False)
-    """How many items per input along concat_dim."""
-
-    def __post_init__(self):
-        self._validate_file_pattern()
-
-        # from here on we know there is at most one merge dim and one concat dim
-        self.concat_dim = self.file_pattern.concat_dims[0]
-        self.cache_metadata = any(
-            [v is None for v in self.file_pattern.concat_sequence_lens.values()]
-        )
-        self.nitems_per_input = self.file_pattern.nitems_per_input[self.concat_dim]
-
-        if self.file_pattern.file_type == FileType.opendap:
-            if self.cache_inputs:
-                raise ValueError("Can't cache opendap inputs.")
-            else:
-                self.cache_inputs = False
-            if self.open_input_with_kerchunk:
-                raise ValueError("Can't generate references on opendap inputs")
-            if "engine" in self.xarray_open_kwargs:
-                if self.xarray_open_kwargs["engine"] != "netcdf4":
-                    raise ValueError(
-                        "Opendap inputs only work with `xarray_open_kwargs['engine'] == 'netcdf4'`"
-                    )
-            else:
-                new_kw = self.xarray_open_kwargs.copy()
-                new_kw["engine"] = "netcdf4"
-                self.xarray_open_kwargs = new_kw
-        elif self.cache_inputs is None:
-            self.cache_inputs = True  # old defult
-
-        self._validate_input_and_chunk_keys()
-
-        # set concat_dim_chunks
-        target_concat_dim_chunks = self.target_chunks.get(self.concat_dim)
-        if (self.nitems_per_input is None) and (target_concat_dim_chunks is None):
-            raise ValueError(
-                "Unable to determine target chunks. Please specify either "
-                "`target_chunks` or `nitems_per_input`"
-            )
-        elif target_concat_dim_chunks:
-            self.concat_dim_chunks = target_concat_dim_chunks
-        else:
-            self.concat_dim_chunks = self.nitems_per_input * self.inputs_per_chunk
-
-    def _validate_file_pattern(self):
-        if len(self.file_pattern.merge_dims) > 1:
-            raise NotImplementedError("This Recipe class can't handle more than one merge dim.")
-        if len(self.file_pattern.concat_dims) > 1:
-            raise NotImplementedError("This Recipe class can't handle more than one concat dim.")
-
-    def _validate_input_and_chunk_keys(self):
-        all_input_keys = set(self.iter_inputs())
-        ninputs = self.file_pattern.dims[self.file_pattern.concat_dims[0]]
-        all_inputs_for_chunks = set(
-            list(
-                chain(
-                    *(
-                        inputs_for_chunk(chunk_key, self.inputs_per_chunk, ninputs)
-                        for chunk_key in self.iter_chunks()
-                    )
-                )
-            )
-        )
-        if all_input_keys != all_inputs_for_chunks:
-            chunk_key = next(iter(self.iter_chunks()))
-            print("First chunk", chunk_key)
-            print("Inputs_for_chunk", inputs_for_chunk(chunk_key, self.inputs_per_chunk, ninputs))
-            raise ValueError("Inputs and chunks are inconsistent")
-
-    def iter_inputs(self) -> Iterator[InputKey]:
-        yield from self.file_pattern
-
-    def iter_chunks(self) -> Iterator[ChunkKey]:
-        for input_key in self.iter_inputs():
-            concat_dims = [
-                dim_idx for dim_idx in input_key if dim_idx.operation == CombineOp.CONCAT
-            ]
-            assert len(concat_dims) == 1
-            concat_dim = concat_dims[0]
-            input_concat_index = concat_dim.index
-            if input_concat_index % self.inputs_per_chunk > 0:
-                continue  # don't emit a chunk
-            chunk_concat_index = input_concat_index // self.inputs_per_chunk
-            chunk_sequence_len = ceil(concat_dim.sequence_len / self.inputs_per_chunk)
-            chunk_concat_dim = replace(
-                concat_dim, index=chunk_concat_index, sequence_len=chunk_sequence_len
-            )
-            chunk_key_base = [
-                chunk_concat_dim if dim_idx.operation == CombineOp.CONCAT else dim_idx
-                for dim_idx in input_key
-            ]
-            if len(self.subset_inputs) == 0:
-                yield Index(chunk_key_base)
-                # no subsets
-                continue
-            subset_iterators = [range(v) for k, v in self.subset_inputs.items()]
-            for i in product(*subset_iterators):
-                # TODO: remove redundant name
-                subset_dims = [
-                    DimIndex(*args, CombineOp.SUBSET)
-                    for args in zip(self.subset_inputs.keys(), i, self.subset_inputs.values())
-                ]
-                yield Index((chunk_key_base + subset_dims))
diff --git a/pangeo_forge_recipes/transforms.py b/pangeo_forge_recipes/transforms.py
index 673637a6..63939299 100644
--- a/pangeo_forge_recipes/transforms.py
+++ b/pangeo_forge_recipes/transforms.py
@@ -4,12 +4,13 @@
 from dataclasses import dataclass, field
 
 # from functools import wraps
-from typing import Optional, Tuple, TypeVar
+from typing import List, Optional, Tuple, TypeVar
 
 import apache_beam as beam
 
+from .combiners import CombineXarraySchemas
 from .openers import open_url, open_with_xarray
-from .patterns import FileType, Index
+from .patterns import DimKey, FileType, Index
 from .storage import CacheFSSpecTarget
 
 logger = logging.getLogger(__name__)
@@ -118,3 +119,48 @@ def expand(self, pcoll):
             copy_to_local=self.copy_to_local,
             xarray_open_kwargs=self.xarray_open_kwargs,
         )
+
+
+def _nest_dim(item: Indexed[T], dim_key: DimKey) -> Indexed[Indexed[T]]:
+    index, value = item
+    inner_index = Index({dim_key: index[dim_key]})
+    outer_index = Index({dk: index[dk] for dk in index if dk != dim_key})
+    return outer_index, (inner_index, value)
+
+
+@dataclass
+class _NestDim(beam.PTransform):
+    """Prepare a collection for grouping by transforming an Index into a nested
+    Tuple of Indexes.
+
+    :param dim_key: The dimension to nest
+    """
+
+    dim_key: DimKey
+
+    def expand(self, pcoll):
+        return pcoll | beam.Map(_nest_dim, dim_key=self.dim_key)
+
+
+@dataclass
+class DetermineSchema(beam.PTransform):
+    """Combine many Dataset schemas into a single schema along multiple dimensions.
+    This is a reduction that produces a singleton PCollection.
+
+    :param combine_dims: The dimensions to combine
+    """
+
+    combine_dims: List[DimKey]
+
+    def expand(self, pcoll: beam.PCollection) -> beam.PCollection:
+        cdims = self.combine_dims.copy()
+        while len(cdims) > 0:
+            last_dim = cdims.pop()
+            if len(cdims) == 0:
+                # at this point, we should have a 1D index as our key
+                pcoll = pcoll | beam.CombineGlobally(CombineXarraySchemas(last_dim))
+            else:
+                pcoll = (
+                    pcoll | _NestDim(last_dim) | beam.CombinePerKey(CombineXarraySchemas(last_dim))
+                )
+        return pcoll
diff --git a/setup.cfg b/setup.cfg
index ac6165d9..82a62a6a 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -66,6 +66,7 @@ line_length=100
 log_cli = False
 timeout = 30
 timeout_method = signal
+addopts = --ignore-glob=tests/recipe_tests/*
 markers =
     no_executor: Tests that do not use an executor fixture.
     executor_function: Tests that use the Python function executor.


Test patch:
diff --git a/tests/test_aggregation.py b/tests/test_aggregation.py
new file mode 100644
index 00000000..7fc8e74d
--- /dev/null
+++ b/tests/test_aggregation.py
@@ -0,0 +1,118 @@
+import numpy as np
+import pandas as pd
+import pytest
+import xarray as xr
+
+from pangeo_forge_recipes.aggregation import DatasetCombineError, XarrayCombineAccumulator
+
+
+def make_ds(nt=10):
+    """Return a synthetic random xarray dataset."""
+    np.random.seed(2)
+    # TODO: change nt to 11 in order to catch the edge case where
+    # items_per_input does not evenly divide the length of the sequence dimension
+    ny, nx = 18, 36
+    time = pd.date_range(start="2010-01-01", periods=nt, freq="D")
+    lon = (np.arange(nx) + 0.5) * 360 / nx
+    lon_attrs = {"units": "degrees_east", "long_name": "longitude"}
+    lat = (np.arange(ny) + 0.5) * 180 / ny
+    lat_attrs = {"units": "degrees_north", "long_name": "latitude"}
+    foo = np.random.rand(nt, ny, nx)
+    foo_attrs = {"long_name": "Fantastic Foo"}
+    # make sure things work with heterogenous data types
+    bar = np.random.randint(0, 10, size=(nt, ny, nx))
+    bar_attrs = {"long_name": "Beautiful Bar"}
+    dims = ("time", "lat", "lon")
+    ds = xr.Dataset(
+        {"bar": (dims, bar, bar_attrs), "foo": (dims, foo, foo_attrs)},
+        coords={
+            "time": ("time", time),
+            "lat": ("lat", lat, lat_attrs),
+            "lon": ("lon", lon, lon_attrs),
+        },
+        attrs={"conventions": "CF 1.6"},
+    )
+    return ds
+
+
+def test_concat_accumulator():
+    ds = make_ds(nt=3)
+    s = ds.to_dict(data=False)  # expected
+
+    aca = XarrayCombineAccumulator(concat_dim="time")
+    aca.add_input(s, 0)
+    s1 = s.copy()
+    s1["chunks"] = {"time": {0: 3}}
+    assert aca.schema == s1
+
+    assert "chunks" not in s
+    aca.add_input(s, 1)
+    s2 = make_ds(nt=6).to_dict(data=False)
+    s2["chunks"] = {"time": {0: 3, 1: 3}}
+    assert aca.schema == s2
+
+    aca2 = XarrayCombineAccumulator(concat_dim="time")
+    aca2.add_input(s, 2)
+    aca_sum = aca + aca2
+    s3 = make_ds(nt=9).to_dict(data=False)
+    s3["chunks"] = {"time": {0: 3, 1: 3, 2: 3}}
+    assert aca_sum.schema == s3
+
+    # now modify attrs and see that fields get dropped correctly
+    ds2 = make_ds(nt=4)
+    ds2.attrs["conventions"] = "wack conventions"
+    ds2.bar.attrs["long_name"] = "nonsense name"
+    aca_sum.add_input(ds2.to_dict(data=False), 3)
+    ds_expected = make_ds(nt=13)
+    del ds_expected.attrs["conventions"]
+    del ds_expected.bar.attrs["long_name"]
+    s4 = ds_expected.to_dict(data=False)
+    s4["chunks"] = {"time": {0: 3, 1: 3, 2: 3, 3: 4}}
+    assert aca_sum.schema == s4
+
+    # make sure we can add in different order
+    aca_sum.add_input(make_ds(nt=1).to_dict(data=False), 5)
+    aca_sum.add_input(make_ds(nt=2).to_dict(data=False), 4)
+    time_chunks = {0: 3, 1: 3, 2: 3, 3: 4, 4: 2, 5: 1}
+    assert aca_sum.schema["chunks"]["time"] == time_chunks
+
+    # now start checking errors
+    ds3 = make_ds(nt=1).isel(lon=slice(1, None))
+    with pytest.raises(DatasetCombineError, match="different sizes"):
+        aca_sum.add_input(ds3.to_dict(data=False), 6)
+
+    with pytest.raises(DatasetCombineError, match="overlapping keys"):
+        aca_sum.add_input(s, 5)
+
+    # now pretend we are concatenating along a new dimension
+    s_time_concat = aca_sum.schema.copy()
+    aca2 = XarrayCombineAccumulator(concat_dim="lon")
+    aca2.add_input(s_time_concat, 0)
+    aca2.add_input(s_time_concat, 1)
+    assert aca2.schema["chunks"]["time"] == time_chunks
+    assert aca2.schema["chunks"]["lon"] == {0: 36, 1: 36}
+    assert aca2.schema["dims"] == {"time": 16, "lat": 18, "lon": 72}
+
+    # check error if we try to concat something with incompatible chunks
+    bad_schema = s_time_concat.copy()
+    bad_schema["chunks"] = s_time_concat["chunks"].copy()
+    bad_schema["chunks"]["time"] = {0: 3, 1: 3, 2: 3, 3: 4, 4: 1, 5: 2}
+    with pytest.raises(DatasetCombineError, match="Non concat_dim chunks must be the same"):
+        aca2.add_input(bad_schema, 2)
+
+    # finally merge; make copy of data with upper case data var names
+    modified_schema = s_time_concat.copy()
+    modified_schema["data_vars"] = s_time_concat["data_vars"].copy()
+    for dvar in s_time_concat["data_vars"]:
+        del modified_schema["data_vars"][dvar]
+        modified_schema["data_vars"][dvar.upper()] = s_time_concat["data_vars"][dvar]
+
+    merge_accumulator = XarrayCombineAccumulator()
+    merge_accumulator.add_input(s_time_concat, 0)
+    merge_accumulator.add_input(modified_schema, 0)
+    assert (
+        merge_accumulator.schema["data_vars"]["foo"] == merge_accumulator.schema["data_vars"]["FOO"]
+    )
+    assert (
+        merge_accumulator.schema["data_vars"]["bar"] == merge_accumulator.schema["data_vars"]["BAR"]
+    )
diff --git a/tests/test_combiners.py b/tests/test_combiners.py
new file mode 100644
index 00000000..3b66008c
--- /dev/null
+++ b/tests/test_combiners.py
@@ -0,0 +1,171 @@
+import apache_beam as beam
+import pytest
+import xarray as xr
+from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.testing.test_pipeline import TestPipeline
+
+# from apache_beam.testing.util import assert_that, equal_to
+from apache_beam.testing.util import assert_that
+from pytest_lazyfixture import lazy_fixture
+
+from pangeo_forge_recipes.combiners import CombineXarraySchemas
+from pangeo_forge_recipes.patterns import CombineOp, DimKey, FilePattern, Index
+from pangeo_forge_recipes.transforms import DetermineSchema, _NestDim
+
+
+@pytest.fixture
+def pipeline():
+    # Runtime type checking doesn't play well with our Combiner
+    # https://github.com/apache/beam/blob/3cddfaf58c69acc624dac16df10357a78ececf59/sdks/python/apache_beam/transforms/core.py#L2505-L2509
+    options = PipelineOptions(runtime_type_check=False)
+    with TestPipeline(options=options) as p:
+        yield p
+
+
+def _get_schema(path):
+    ds = xr.open_dataset(path)
+    return ds.to_dict(data=False)
+
+
+def _expected_schema(pattern):
+    expected_ds = xr.open_mfdataset(item[1] for item in pattern.items())
+    expected_schema = expected_ds.to_dict(data=False)
+    expected_schema["chunks"] = {
+        "time": {pos: v for pos, v in enumerate(expected_ds.chunks["time"])}
+    }
+    return expected_schema
+
+
+# TODO: make this fixture leaner; don't need netcdf3 and netcdf4
+@pytest.fixture(
+    scope="module",
+    params=[
+        lazy_fixture("netcdf_local_file_pattern_sequential"),
+    ],
+    ids=["sequential"],
+)
+def schema_pcoll_concat(request):
+    pattern = request.param
+    expected_schema = _expected_schema(pattern)
+    return (
+        pattern,
+        expected_schema,
+        beam.Create(((key, _get_schema(url)) for key, url in pattern.items())),
+    )
+
+
+@pytest.fixture(
+    scope="module",
+    params=[
+        lazy_fixture("netcdf_local_file_pattern_sequential_multivariable"),
+    ],
+    ids=["sequential_multivariable"],
+)
+def schema_pcoll_concat_merge(request):
+    pattern = request.param
+    expected_schema = _expected_schema(pattern)
+    return (
+        pattern,
+        expected_schema,
+        beam.Create(((key, _get_schema(url)) for key, url in pattern.items())),
+    )
+
+
+def _get_concat_dim(pattern):
+    cdims = pattern.concat_dims
+    assert len(cdims) == 1, "Only one concat_dim allowed for now"
+    return cdims[0]
+
+
+def _strip_keys(item):
+    return item[1]
+
+
+def has_correct_schema(expected_schema):
+    def _check_results(actual):
+        assert len(actual) == 1
+        schema = actual[0]
+        assert schema == expected_schema
+
+    return _check_results
+
+
+def test_CombineXarraySchemas_concat_1D(schema_pcoll_concat, pipeline):
+    pattern, expected_schema, pcoll = schema_pcoll_concat
+    concat_dim = _get_concat_dim(pattern)
+
+    with pipeline as p:
+        input = p | pcoll
+        output = input | beam.CombineGlobally(
+            CombineXarraySchemas(DimKey(name=concat_dim, operation=CombineOp.CONCAT))
+        )
+        assert_that(output, has_correct_schema(expected_schema))
+
+
+# TODO: maybe remove this test. It testing an implementation detail
+def test_NestDim(schema_pcoll_concat_merge, pipeline):
+    pattern, _, pcoll = schema_pcoll_concat_merge
+    pattern_merge_only = FilePattern(
+        pattern.format_function,
+        *[cdim for cdim in pattern.combine_dims if cdim.operation == CombineOp.MERGE]
+    )
+    merge_only_indexes = list(pattern_merge_only)
+    pattern_concat_only = FilePattern(
+        pattern.format_function,
+        *[cdim for cdim in pattern.combine_dims if cdim.operation == CombineOp.CONCAT]
+    )
+    concat_only_indexes = list(pattern_concat_only)
+
+    def check_key(expected_outer_indexes, expected_inner_indexes):
+        def _check(actual):
+            outer_indexes = [item[0] for item in actual]
+            assert outer_indexes == expected_outer_indexes
+            for outer_index, group in actual:
+                assert isinstance(outer_index, Index)
+                inner_indexes = [item[0] for item in group]
+                assert inner_indexes == expected_inner_indexes
+                for idx in inner_indexes:
+                    assert isinstance(idx, Index)
+
+        return _check
+
+    with pipeline as p:
+        input = p | pcoll
+        group1 = (
+            input
+            | "Nest CONCAT" >> _NestDim(DimKey("time", CombineOp.CONCAT))
+            | "Groupby CONCAT" >> beam.GroupByKey()
+        )
+        group2 = (
+            input
+            | "Nest MERGE" >> _NestDim(DimKey("variable", CombineOp.MERGE))
+            | "Groupy MERGE" >> beam.GroupByKey()
+        )
+        assert_that(group1, check_key(merge_only_indexes, concat_only_indexes), label="merge")
+        assert_that(group2, check_key(concat_only_indexes, merge_only_indexes), label="concat")
+
+
+def test_DetermineSchema_concat_1D(schema_pcoll_concat, pipeline):
+    pattern, expected_schema, pcoll = schema_pcoll_concat
+    concat_dim = _get_concat_dim(pattern)
+
+    with pipeline as p:
+        input = p | pcoll
+        output = input | DetermineSchema([DimKey(name=concat_dim, operation=CombineOp.CONCAT)])
+        assert_that(output, has_correct_schema(expected_schema))
+
+
+_dimkeys = [
+    DimKey("time", operation=CombineOp.CONCAT),
+    DimKey("variable", operation=CombineOp.MERGE),
+]
+
+
+@pytest.mark.parametrize("dimkeys", [_dimkeys, _dimkeys[::-1]], ids=["concat_first", "merge_first"])
+def test_DetermineSchema_concat_merge(dimkeys, schema_pcoll_concat_merge, pipeline):
+    pattern, expected_schema, pcoll = schema_pcoll_concat_merge
+
+    with pipeline as p:
+        input = p | pcoll
+        output = input | DetermineSchema(dimkeys)
+        assert_that(output, has_correct_schema(expected_schema))
diff --git a/tests/test_patterns.py b/tests/test_patterns.py
index 62a286fe..ea73a45a 100644
--- a/tests/test_patterns.py
+++ b/tests/test_patterns.py
@@ -5,7 +5,7 @@
 from pangeo_forge_recipes.patterns import (
     CombineOp,
     ConcatDim,
-    DimIndex,
+    DimVal,
     FilePattern,
     FileType,
     MergeDim,
@@ -81,8 +81,9 @@ def test_pattern_from_file_sequence():
     assert fp.concat_dims == ["time"]
     assert fp.nitems_per_input == {"time": None}
     assert fp.concat_sequence_lens == {"time": None}
-    for key in fp:
-        assert fp[key] == file_sequence[sorted(key)[0].index]
+    for index in fp:
+        dimval = next(iter(index.values()))
+        assert fp[index] == file_sequence[dimval.position]
 
 
 @pytest.mark.parametrize("pickle", [False, True])
@@ -121,15 +122,13 @@ def test_file_pattern_concat_merge(runtime_secrets, pickle, concat_merge_pattern
     assert fp.concat_sequence_lens == {"time": None}
     assert len(list(fp)) == 6
     for key in fp:
-        for k in key:
-            if k.name == "time":
-                assert k.operation == CombineOp.CONCAT
-                assert k.sequence_len == 3
-                time_val = times[k.index]
-            if k.name == "variable":
-                assert k.operation == CombineOp.MERGE
-                assert k.sequence_len == 2
-                variable_val = varnames[k.index]
+        for dimkey, dimval in key.items():
+            if dimkey.name == "time":
+                assert dimkey.operation == CombineOp.CONCAT
+                time_val = times[dimval.position]
+            if dimkey.name == "variable":
+                assert dimkey.operation == CombineOp.MERGE
+                variable_val = varnames[dimval.position]
         expected_fname = format_function(time=time_val, variable=variable_val)
         assert fp[key] == expected_fname
 
@@ -201,24 +200,11 @@ def test_setting_file_types(file_type_value):
 
 
 @pytest.mark.parametrize(
-    "index,start,stop",
+    "position,start,stop",
     [(0, 0, 2), (1, 2, 4), (2, 4, 7), (3, 7, 9), (4, 9, 11)],
 )
-def test_augment_index_with_start_stop(index, start, stop):
-    kw = dict(name="time", index=index, sequence_len=5, operation=CombineOp.CONCAT)
-    di = DimIndex(**kw)
-    expected = DimIndex(start=start, stop=stop, **kw)
-    actual = augment_index_with_start_stop(di, [2, 2, 3, 2, 2])
+def test_augment_index_with_start_stop(position, start, stop):
+    dk = DimVal(position)
+    expected = DimVal(position, start, stop)
+    actual = augment_index_with_start_stop(dk, [2, 2, 3, 2, 2])
     assert actual == expected
-
-
-@pytest.mark.parametrize(
-    "dimindex_sequence_len, passed_sequence_lengths",
-    [(2, [1]), (2, [1, 1, 1])],
-)
-def test_augment_index_with_start_stop_raises(dimindex_sequence_len, passed_sequence_lengths):
-    kw = dict(name="time", index=0, operation=CombineOp.CONCAT)
-    di = DimIndex(sequence_len=dimindex_sequence_len, **kw)
-
-    with pytest.raises(AssertionError):
-        augment_index_with_start_stop(di, passed_sequence_lengths)
diff --git a/tests/test_serialization.py b/tests/test_serialization.py
index 231bdae1..cf43ca73 100644
--- a/tests/test_serialization.py
+++ b/tests/test_serialization.py
@@ -1,15 +1,15 @@
-from dataclasses import asdict, dataclass, field
+from dataclasses import asdict, dataclass
 from datetime import datetime, timedelta
-from typing import Callable, Optional
+from typing import Callable
 
 import pandas as pd
 import pytest
-from fsspec.implementations.local import LocalFileSystem
 
 from pangeo_forge_recipes.patterns import ConcatDim, FilePattern, FileType, match_pattern_blockchain
-from pangeo_forge_recipes.recipes import HDFReferenceRecipe, XarrayZarrRecipe
 from pangeo_forge_recipes.serialization import dict_to_sha256, either_encode_or_hash
-from pangeo_forge_recipes.storage import FSSpecTarget, StorageConfig
+
+# TODO: revise this test once refactor is farther along
+pytest.skip(allow_module_level=True)
 
 # TODO: revise this test once refactor is farther along
 pytest.skip(allow_module_level=True)
@@ -90,89 +90,6 @@ def test_match_pattern_blockchain(base_pattern, end_date, new_pattern_nitems_per
         assert matching_key is None
 
 
-@pytest.mark.parametrize("recipe_cls", [XarrayZarrRecipe, HDFReferenceRecipe])
-def test_recipe_sha256_hash_exclude(base_pattern, recipe_cls, tmpdir_factory):
-    recipe_0 = recipe_cls(base_pattern)
-    recipe_1 = recipe_cls(base_pattern)
-
-    assert recipe_0.sha256() == recipe_1.sha256()
-
-    local_fs = LocalFileSystem()
-    custom_target_path = tmpdir_factory.mktemp("custom_target")
-    custom_storage_config = StorageConfig(target=FSSpecTarget(local_fs, custom_target_path))
-    recipe_1.storage_config = custom_storage_config
-
-    assert recipe_0.sha256() == recipe_1.sha256()
-
-
-@pytest.mark.parametrize(
-    "kwargs",
-    [
-        ({}, {}),
-        ({}, dict(target_chunks={"time": 1})),
-        (dict(target_chunks={"time": 1}), dict(target_chunks={"time": 1})),
-        (dict(target_chunks={"time": 1}), dict(target_chunks={"time": 2})),
-        (dict(subset_inputs={"time": 2}), dict(target_chunks={"time": 2})),
-    ],
-)
-def test_xarray_zarr_sha265(pattern_pair, kwargs):
-    recipe_0 = XarrayZarrRecipe(pattern_pair[0], **kwargs[0])
-    recipe_1 = XarrayZarrRecipe(pattern_pair[1], **kwargs[1])
-
-    if pattern_pair[0] == pattern_pair[1] and kwargs[0] == kwargs[1]:
-        assert recipe_0.sha256() == recipe_1.sha256()
-    else:
-        assert recipe_0.sha256() != recipe_1.sha256()
-
-
-@pytest.mark.parametrize(
-    "kwargs",
-    [
-        ({}, {}),
-        ({}, dict(output_json_fname="custom_name.json")),
-        (dict(output_json_fname="custom_name.json"), dict(output_json_fname="custom_name.json")),
-        (dict(netcdf_storage_options={"anon": True}), dict(output_json_fname="custom_name.json")),
-    ],
-)
-def test_kerchunk_sha265(pattern_pair, kwargs):
-    recipe_0 = HDFReferenceRecipe(pattern_pair[0], **kwargs[0])
-    recipe_1 = HDFReferenceRecipe(pattern_pair[1], **kwargs[1])
-
-    if pattern_pair[0] == pattern_pair[1] and kwargs[0] == kwargs[1]:
-        assert recipe_0.sha256() == recipe_1.sha256()
-    else:
-        assert recipe_0.sha256() != recipe_1.sha256()
-
-
-@pytest.mark.parametrize("cls", [XarrayZarrRecipe, HDFReferenceRecipe])
-@pytest.mark.parametrize(
-    "kwargs",
-    [
-        {},
-        {"new_optional_str": "hello"},
-        {"new_dict": dict(a=1)},
-        {"new_list": [1, 2, 3]},
-    ],
-)
-def test_additional_fields(base_pattern, cls, kwargs):
-    # simulates a new release in which new fields are added; because we drop empty fields from
-    # the hash calculation, backwards compatibility is preserved as long as new fields are unset
-
-    @dataclass
-    class NewRelease(cls):
-        new_optional_str: Optional[str] = None
-        new_dict: dict = field(default_factory=dict)
-        new_list: list = field(default_factory=list)
-
-    old_release_obj = cls(base_pattern)
-    new_release_obj = NewRelease(base_pattern, **kwargs)
-
-    if not kwargs:
-        assert old_release_obj.sha256() == new_release_obj.sha256()
-    else:
-        assert old_release_obj.sha256() != new_release_obj.sha256()
-
-
 def test_either_encode_or_hash_raises():
     def f():
         pass