diff --git a/.github/workflows/testing.yml b/.github/workflows/testing.yml
index 0b9ba26d6..d559464c7 100644
--- a/.github/workflows/testing.yml
+++ b/.github/workflows/testing.yml
@@ -16,7 +16,7 @@ jobs:
         - 27017:27017
     strategy:
       matrix:
-        python-version: [3.7, 3.8, 3.9]
+        python-version: ["3.8", "3.9", "3.10"]
       fail-fast: false
     steps:
 
diff --git a/databroker/experimental/__init__.py b/databroker/experimental/__init__.py
deleted file mode 100644
index e69de29bb..000000000
diff --git a/databroker/experimental/schemas.py b/databroker/experimental/schemas.py
deleted file mode 100644
index 4549fd283..000000000
--- a/databroker/experimental/schemas.py
+++ /dev/null
@@ -1,112 +0,0 @@
-from typing import Dict, List, Optional, Union
-from datetime import datetime
-
-import pydantic
-import pydantic.generics
-
-from tiled.server.pydantic_array import ArrayStructure
-from tiled.server.pydantic_dataframe import DataFrameStructure
-from tiled.server.pydantic_sparse import SparseStructure
-from tiled.structures.core import StructureFamily
-
-from tiled.server.schemas import ReferenceDocument, Spec
-
-# Map structure family to the associated
-# structure model. This is used by the validator.
-structure_association = {
-    StructureFamily.array: ArrayStructure,
-    StructureFamily.dataframe: DataFrameStructure,
-    StructureFamily.sparse: SparseStructure,
-    # StructureFamily.node
-    # ...
-}
-
-
-class BaseDocument(pydantic.BaseModel):
-    key: str
-    metadata: Dict
-    specs: List[Spec]
-    references: List[ReferenceDocument]
-    updated_at: datetime
-
-
-class Document(BaseDocument):
-    structure_family: StructureFamily
-    structure: Union[ArrayStructure, DataFrameStructure, SparseStructure]
-    mimetype: str
-    created_at: datetime
-    data_blob: Optional[bytes]
-    data_url: Optional[pydantic.AnyUrl]
-
-    @pydantic.root_validator
-    def validate_structure_matches_structure_family(cls, values):
-        actual_structure = values.get("structure")
-        # Given the structure_family, we know what the structure type should be.
-        expected_structure_type = structure_association[values.get("structure_family")]
-        if expected_structure_type == StructureFamily.node:
-            raise Exception(
-                f"{expected_structure_type} is not currently supported as a writable structure"
-            )
-        elif not isinstance(actual_structure, expected_structure_type):
-            raise Exception(
-                "The expected structure type does not match the received structure type"
-            )
-        return values
-
-    @pydantic.validator("mimetype")
-    def is_mime_type(cls, v):
-        m_type, _, _ = v.partition("/")
-        mime_type_list = set(
-            [
-                "application",
-                "audio",
-                "font",
-                "example",
-                "image",
-                "message",
-                "model",
-                "multipart",
-                "text",
-                "video",
-            ]
-        )
-
-        if m_type not in mime_type_list:
-            raise ValueError(f"{m_type} is not a valid mime type")
-        return v
-
-    @pydantic.root_validator
-    def check_data_source(cls, values):
-        # Making them optional and setting default values might help to meet these conditions
-        # with the current data types without getting any conflicts
-        if values.get("data_blob") is not None and values.get("data_url") is not None:
-            raise ValueError(
-                "Not Valid: data_blob and data_url contain values. Use just one"
-            )
-        return values
-
-
-class DocumentRevision(BaseDocument):
-    revision: int
-
-    @classmethod
-    def from_document(cls, document, revision):
-        return cls(
-            key=document.key,
-            metadata=document.metadata,
-            specs=document.specs,
-            references=document.references,
-            updated_at=document.updated_at,
-            revision=revision,
-        )
-
-    @classmethod
-    def from_json(cls, json_doc):
-        return cls(
-            key=json_doc["key"],
-            metadata=json_doc["metadata"],
-            specs=json_doc["specs"],
-            references=json_doc["references"],
-            updated_at=json_doc["updated_at"],
-            revision=json_doc["revision"],
-        )
diff --git a/databroker/experimental/server_ext.py b/databroker/experimental/server_ext.py
deleted file mode 100644
index 27ba27f96..000000000
--- a/databroker/experimental/server_ext.py
+++ /dev/null
@@ -1,810 +0,0 @@
-import builtins
-import collections.abc
-import itertools
-import os
-import shutil
-import uuid
-import time
-from pathlib import Path
-from datetime import datetime, timezone
-
-import dask.dataframe
-import numpy
-import pymongo
-import sparse
-import zarr
-import zarr.storage
-
-from tiled.adapters.array import slice_and_shape_from_block_and_chunks
-from tiled.adapters.array import ArrayAdapter
-from tiled.adapters.dataframe import DataFrameAdapter
-from tiled.adapters.mapping import MapAdapter
-from tiled.adapters.utils import IndexersMixin, tree_repr
-from tiled.iterviews import KeysView, ItemsView, ValuesView
-from tiled.queries import (
-    Contains,
-    Comparison,
-    Eq,
-    FullText,
-    In,
-    KeysFilter,
-    NotEq,
-    NotIn,
-    Operator,
-    Regex,
-)
-from tiled.query_registration import QueryTranslationRegistry
-
-from tiled.structures.core import Spec, StructureFamily
-from tiled.structures.array import ArrayStructure, ArrayMacroStructure, BuiltinDtype
-from tiled.structures.dataframe import (
-    DataFrameStructure,
-    DataFrameMacroStructure,
-    DataFrameMicroStructure,
-)
-from tiled.structures.sparse import COOStructure
-from tiled.serialization.dataframe import deserialize_arrow
-
-from tiled.utils import APACHE_ARROW_FILE_MIME_TYPE, UNCHANGED, OneShotCachedMap
-
-from .schemas import Document, DocumentRevision
-
-from sys import modules, platform
-
-
-class Revisions:
-    def __init__(self, collection, key):
-        self._collection = collection
-        self._key = key
-
-    def __len__(self):
-        return self._collection.count_documents({"key": self._key})
-
-    def __getitem__(self, item_):
-        offset = item_.start
-        limit = item_.stop - offset
-        return list(
-            self._collection.find({"key": self._key})
-            .sort("revision", 1)
-            .skip(offset)
-            .limit(limit)
-        )
-
-    def delete_revision(self, n):
-        self._collection.delete_one({"key": self._key, "revision": n})
-
-    def delete_all(self):
-        self._collection.delete_many({"key": self._key})
-
-    def last(self):
-        return self._collection.find_one({"key": self._key}, sort=[("revision", -1)])
-
-    def add_document(self, document):
-        return self._collection.insert_one(document.dict())
-
-
-class WritingArrayAdapter:
-    structure_family = "array"
-
-    def __init__(self, database, key):
-        self.collection = database["nodes"]
-        self.revisions = Revisions(database["revisions"], key)
-        self.key = key
-        self.deadline = 0
-        assert self.doc.data_blob is None  # not implemented
-        self.array = zarr.open_array(str(safe_path(self.doc.data_url.path)), "r+")
-
-    @classmethod
-    def new(cls, database, doc):
-        # Zarr requires evently-sized chunks within each dimension.
-        # Use the first chunk along each dimension.
-        chunks = tuple(dim[0] for dim in doc.structure.macro.chunks)
-        shape = tuple(dim[0] * len(dim) for dim in doc.structure.macro.chunks)
-        storage = zarr.storage.DirectoryStore(str(safe_path(doc.data_url.path)))
-        zarr.storage.init_array(
-            storage,
-            shape=shape,
-            chunks=chunks,
-            dtype=doc.structure.micro.to_numpy_dtype(),
-        )
-        return cls(database, doc.key)
-
-    @property
-    def doc(self):
-        now = time.monotonic()
-        if now > self.deadline:
-            self._doc = Document(**self.collection.find_one({"key": self.key}))
-            self.deadline = now + 0.1  # In seconds
-
-        return self._doc
-
-    @property
-    def structure(self):
-        # Convert pydantic implementation to dataclass implemenetation
-        # expected by server.
-        return ArrayStructure(**self.doc.structure.dict())
-
-    @property
-    def metadata(self):
-        return self.doc.metadata
-
-    @property
-    def specs(self):
-        # Convert from pydantic to dataclass.
-        return [Spec(**dict(spec)) for spec in self.doc.specs]
-
-    @property
-    def references(self):
-        return self.doc.references
-
-    def read(self, slice=None):
-        # Trim overflow because Zarr always has equal-sized chunks.
-        arr = self.array[
-            tuple(builtins.slice(0, dim) for dim in self.doc.structure.macro.shape)
-        ]
-        if slice is not None:
-            arr = arr[slice]
-        return arr
-
-    def read_block(self, block, slice=None):
-        # Trim overflow because Zarr always has equal-sized chunks.
-        slice_, _ = slice_and_shape_from_block_and_chunks(
-            block, self.doc.structure.macro.chunks
-        )
-        # Slice the block out of the whole array.
-        arr = self.array[slice_]
-        # And then maybe slice *within* the block.
-        if slice is not None:
-            arr = arr[slice]
-        return arr
-
-    def microstructure(self):
-        return BuiltinDtype(**self.doc.structure.micro.dict())
-
-    def macrostructure(self):
-        return ArrayMacroStructure(**self.doc.structure.macro.dict())
-
-    def put_data(self, body, block=None):
-        # Organize files into subdirectories with the first two
-        # characters of the key to avoid one giant directory.
-        if block:
-            slice_, shape = slice_and_shape_from_block_and_chunks(
-                block, self.doc.structure.macro.chunks
-            )
-        else:
-            slice_ = numpy.s_[:]
-            shape = self.doc.structure.macro.shape
-        array = numpy.frombuffer(
-            body, dtype=self.doc.structure.micro.to_numpy_dtype()
-        ).reshape(shape)
-        self.array[slice_] = array
-
-    def put_metadata(self, metadata, specs, references):
-        last_revision_doc = self.revisions.last()
-        if last_revision_doc is not None:
-            revision = int(last_revision_doc["revision"]) + 1
-        else:
-            revision = 0
-
-        validated_revision = DocumentRevision.from_document(self.doc, revision)
-
-        self.revisions.add_document(validated_revision)
-        updated_at = datetime.now(tz=timezone.utc)
-
-        to_set = {"updated_at": updated_at}
-
-        if metadata is not None:
-            to_set["metadata"] = metadata
-
-        if specs is not None:
-            to_set["specs"] = [spec.dict() for spec in specs]
-
-        if references is not None:
-            references_dict = [item.dict() for item in references]
-            to_set["references"] = references_dict
-
-        result = self.collection.update_one(
-            {"key": self.key},
-            {"$set": to_set},
-        )
-
-        if result.matched_count != result.modified_count:
-            raise RuntimeError("Error while writing to database")
-
-    def delete(self):
-        shutil.rmtree(safe_path(self.doc.data_url.path))
-        result = self.collection.delete_one({"key": self.key})
-        assert result.deleted_count == 1
-        self.revisions.delete_all()
-
-
-class WritingDataFrameAdapter:
-    structure_family = "dataframe"
-
-    def __init__(self, database, key):
-        self.collection = database["nodes"]
-        self.revisions = Revisions(database["revisions"], key)
-        self.key = key
-        self.deadline = 0
-        assert self.doc.data_blob is None  # not implemented
-
-    @property
-    def dataframe_adapter(self):
-        return DataFrameAdapter.from_dask_dataframe(
-            dask.dataframe.read_parquet(safe_path(self.doc.data_url.path))
-        )
-
-    @classmethod
-    def new(cls, database, doc):
-        safe_path(doc.data_url.path).mkdir(parents=True)
-        return cls(database, doc.key)
-
-    @property
-    def doc(self):
-        now = time.monotonic()
-        if now > self.deadline:
-            self._doc = Document(
-                **self.collection.find_one({"key": self.key})
-            )  # run query
-            self.deadline = now + 0.1  # In seconds
-
-        return self._doc
-
-    @property
-    def structure(self):
-        # Convert pydantic implementation to dataclass implemenetation
-        # expected by server.
-        return DataFrameStructure(**self.doc.structure.dict())
-
-    @property
-    def metadata(self):
-        return self.doc.metadata
-
-    @property
-    def specs(self):
-        # Convert from pydantic to dataclass.
-        return [Spec(**dict(spec)) for spec in self.doc.specs]
-
-    @property
-    def references(self):
-        return self.doc.references
-
-    def __getitem__(self, key):
-        return ArrayAdapter(self.dataframe_adapter.read([key])[key].values)
-
-    def read(self, *args, **kwargs):
-        return self.dataframe_adapter.read(*args, **kwargs)
-
-    def read_partition(self, *args, **kwargs):
-        return self.dataframe_adapter.read_partition(*args, **kwargs)
-
-    def microstructure(self):
-        return DataFrameMicroStructure(**self.doc.structure.micro.dict())
-
-    def macrostructure(self):
-        return DataFrameMacroStructure(**self.doc.structure.macro.dict())
-
-    def put_data(self, body, partition=0):
-        dataframe = deserialize_arrow(body)
-        dataframe.to_parquet(
-            safe_path(self.doc.data_url.path) / f"partition-{partition}.parquet"
-        )
-
-    def put_metadata(self, metadata, specs, references):
-        last_revision_doc = self.revisions.last()
-        if last_revision_doc is not None:
-            revision = int(last_revision_doc["revision"]) + 1
-        else:
-            revision = 0
-
-        validated_revision = DocumentRevision.from_document(self.doc, revision)
-
-        result = self.revisions.add_document(validated_revision)
-        updated_at = datetime.now(tz=timezone.utc)
-
-        to_set = {"updated_at": updated_at}
-
-        if metadata is not None:
-            to_set["metadata"] = metadata
-
-        if specs is not None:
-            to_set["specs"] = [spec.dict() for spec in specs]
-
-        if references is not None:
-            references_dict = [item.dict() for item in references]
-            to_set["references"] = references_dict
-
-        result = self.collection.update_one(
-            {"key": self.key},
-            {"$set": to_set},
-        )
-
-        if result.matched_count != result.modified_count:
-            raise RuntimeError("Error while writing to database")
-
-    def delete(self):
-        shutil.rmtree(safe_path(self.doc.data_url.path))
-        result = self.collection.delete_one({"key": self.doc.key})
-        assert result.deleted_count == 1
-        self.revisions.delete_all()
-
-
-class WritingCOOAdapter:
-    structure_family = "sparse"
-
-    def __init__(self, database, key):
-        def load(filepath):
-            import pandas
-
-            df = pandas.read_parquet(filepath)
-            coords = df[df.columns[:-1]].values.T
-            data = df["data"].values
-            return coords, data
-
-        self.collection = database["nodes"]
-        self.revisions = Revisions(database["revisions"], key)
-        self.key = key
-        self.deadline = 0
-        assert self.doc.data_blob is None  # not implemented
-        num_blocks = (range(len(n)) for n in self.doc.structure.chunks)
-        directory = safe_path(self.doc.data_url.path)
-        mapping = {}
-        for block in itertools.product(*num_blocks):
-            filepath = directory / f"block-{'.'.join(map(str, block))}.parquet"
-            if filepath.is_file():
-                mapping[block] = lambda filepath=filepath: load(filepath)
-        self.blocks = OneShotCachedMap(mapping)
-
-    @property
-    def doc(self):
-        now = time.monotonic()
-        if now > self.deadline:
-            self._doc = Document(
-                **self.collection.find_one({"key": self.key})
-            )  # run query
-            self.deadline = now + 0.1  # In seconds
-
-        return self._doc
-
-    @classmethod
-    def new(cls, database, doc):
-        safe_path(doc.data_url.path).mkdir(parents=True)
-        return cls(database, doc.key)
-
-    @property
-    def metadata(self):
-        return self.doc.metadata
-
-    @property
-    def specs(self):
-        # Convert from pydantic to dataclass.
-        return [Spec(**dict(spec)) for spec in self.doc.specs]
-
-    @property
-    def references(self):
-        return self.doc.references
-
-    def read_block(self, block, slice=None):
-        coords, data = self.blocks[block]
-        _, shape = slice_and_shape_from_block_and_chunks(
-            block, self.doc.structure.chunks
-        )
-        arr = sparse.COO(data=data[:], coords=coords[:], shape=shape)
-        if slice:
-            arr = arr[slice]
-        return arr
-
-    def read(self, slice=None):
-        all_coords = []
-        all_data = []
-        for block, (coords, data) in self.blocks.items():
-            offsets = []
-            for b, c in zip(block, self.doc.structure.chunks):
-                offset = sum(c[:b])
-                offsets.append(offset)
-            global_coords = coords + [[i] for i in offsets]
-            all_coords.append(global_coords)
-            all_data.append(data)
-        arr = sparse.COO(
-            data=numpy.concatenate(all_data),
-            coords=numpy.concatenate(all_coords, axis=-1),
-            shape=self.doc.structure.shape,
-        )
-        if slice:
-            return arr[slice]
-        return arr
-
-    def structure(self):
-        # Convert pydantic implementation to dataclass implemenetation
-        # expected by server.
-        return COOStructure(**self.doc.structure.dict())
-
-    def put_data(self, body, block=None):
-        if block is None:
-            block = (0,) * len(self.doc.structure.shape)
-        dataframe = deserialize_arrow(body)
-        dataframe.to_parquet(
-            safe_path(self.doc.data_url.path)
-            / f"block-{'.'.join(map(str, block))}.parquet"
-        )
-
-    def put_metadata(self, metadata, specs, references):
-        last_revision_doc = self.revisions.last()
-        if last_revision_doc is not None:
-            revision = int(last_revision_doc["revision"]) + 1
-        else:
-            revision = 0
-
-        validated_revision = DocumentRevision.from_document(self.doc, revision)
-
-        result = self.revisions.add_document(validated_revision)
-        updated_at = datetime.now(tz=timezone.utc)
-
-        to_set = {"updated_at": updated_at}
-
-        if metadata is not None:
-            to_set["metadata"] = metadata
-
-        if specs is not None:
-            to_set["specs"] = [spec.dict() for spec in specs]
-
-        if references is not None:
-            references_dict = [item.dict() for item in references]
-            to_set["references"] = references_dict
-
-        result = self.collection.update_one(
-            {"key": self.key},
-            {"$set": to_set},
-        )
-
-        if result.matched_count != result.modified_count:
-            raise RuntimeError("Error while writing to database")
-
-    def delete(self):
-        shutil.rmtree(safe_path(self.doc.data_url.path))
-        result = self.collection.delete_one({"key": self.doc.key})
-        assert result.deleted_count == 1
-        self.revisions.delete_all()
-
-
-class MongoAdapter(collections.abc.Mapping, IndexersMixin):
-    structure_family = "node"
-    query_registry = QueryTranslationRegistry()
-    register_query = query_registry.register
-
-    def __init__(
-        self,
-        *,
-        database,
-        directory,
-        queries=None,
-        sorting=None,
-        metadata=None,
-        access_policy=None,
-    ):
-        self.database = database
-        self.collection = database["nodes"]
-        self.revision_coll = database["revisions"]
-        self.directory = Path(directory).resolve()
-        if not self.directory.exists():
-            raise ValueError(f"Directory {self.directory} does not exist.")
-        if not self.directory.is_dir():
-            raise ValueError(
-                f"The given directory path {self.directory} is not a directory."
-            )
-        if not os.access(self.directory, os.W_OK):
-            raise ValueError(f"Directory {self.directory} is not writeable.")
-        self.queries = queries or []
-        self.sorting = sorting or [("metadata.scan_id", 1)]
-        self.metadata = metadata or {}
-        self.access_policy = access_policy
-        super().__init__()
-
-    @classmethod
-    def from_uri(cls, uri, directory, **kwargs):
-        """
-        When calling this method, call create_index() from its instance to define the
-        unique indexes in the revision collection
-
-        """
-        if not pymongo.uri_parser.parse_uri(uri)["database"]:
-            raise ValueError(
-                f"Invalid URI: {uri!r} Did you forget to include a database?"
-            )
-        client = pymongo.MongoClient(uri)
-        database = client.get_database()
-        return cls(database=database, directory=directory, **kwargs)
-
-    @classmethod
-    def from_mongomock(cls, directory, **kwargs):
-        import mongomock
-
-        db_name = f"temp-{str(uuid.uuid4())}"
-        mongo_client = mongomock.MongoClient()
-        database = mongo_client[db_name]
-
-        mongo_adapter = cls(database=database, directory=directory, **kwargs)
-        mongo_adapter.create_indexes()
-
-        return mongo_adapter
-
-    def new_variation(
-        self,
-        metadata=UNCHANGED,
-        queries=UNCHANGED,
-        sorting=UNCHANGED,
-        **kwargs,
-    ):
-        if metadata is UNCHANGED:
-            metadata = self.metadata
-        if queries is UNCHANGED:
-            queries = self.queries
-        if sorting is UNCHANGED:
-            sorting = self.sorting
-        return type(self)(
-            database=self.database,
-            directory=self.directory,
-            metadata=metadata,
-            queries=queries,
-            sorting=sorting,
-            access_policy=self.access_policy,
-            **kwargs,
-        )
-
-    def post_metadata(self, metadata, structure_family, structure, specs, references):
-        mime_structure_association = {
-            StructureFamily.array: "application/x-zarr",
-            StructureFamily.dataframe: APACHE_ARROW_FILE_MIME_TYPE,
-            StructureFamily.sparse: APACHE_ARROW_FILE_MIME_TYPE,
-        }
-
-        key = str(uuid.uuid4())
-        created_date = datetime.now(tz=timezone.utc)
-
-        validated_document = Document(
-            key=key,
-            structure_family=structure_family,
-            structure=structure,
-            metadata=metadata,
-            specs=specs,
-            references=references,
-            mimetype=mime_structure_association[structure_family],
-            data_url=f"file://localhost/{self.directory}/{key[:2]}/{key}",
-            created_at=created_date,
-            updated_at=created_date,
-        )
-
-        _adapter_class_by_family[structure_family]
-        self.collection.insert_one(validated_document.dict())
-        _adapter_class_by_family[structure_family].new(
-            self.database, validated_document
-        )
-        return key
-
-    def create_indexes(self):
-        self.collection.create_index([("key", pymongo.ASCENDING)], unique=True)
-
-        self.revision_coll.create_index(
-            [("key", pymongo.ASCENDING), ("revision", pymongo.DESCENDING)], unique=True
-        )
-
-    def _build_mongo_query(self, *queries):
-        combined = self.queries + list(queries)
-        if combined:
-            return {"$and": combined}
-        else:
-            return {}
-
-    def __getitem__(self, key):
-        query = {"key": key}
-        doc = self.collection.find_one(self._build_mongo_query(query), {"_id": False})
-        if doc is None:
-            raise KeyError(key)
-
-        class_ = _adapter_class_by_family[StructureFamily(doc["structure_family"])]
-        return class_(self.database, key)
-
-    def __iter__(self):
-        # TODO Apply pagination, as we do in Databroker.
-        for doc in list(
-            self.collection.find(
-                self._build_mongo_query({"data_url": {"$ne": None}}),
-                {"key": True},
-            )
-        ):
-            yield doc["key"]
-
-    def __len__(self):
-        return self.collection.count_documents(
-            self._build_mongo_query({"data_url": {"$ne": None}})
-        )
-
-    def __length_hint__(self):
-        # https://www.python.org/dev/peps/pep-0424/
-        return self.collection.estimated_document_count(
-            self._build_mongo_query({"data_url": {"$ne": None}}),
-        )
-
-    def __repr__(self):
-        # Display up to the first N keys to avoid making a giant service
-        # request. Use _keys_slicer because it is unauthenticated.
-        N = 10
-        return tree_repr(self, self._keys_slice(0, N, direction=1))
-
-    def search(self, query):
-        """
-        Return a MongoAdapter with a subset of the mapping.
-        """
-        return self.query_registry(query, self)
-
-    def get_distinct(self, metadata, structure_families, specs, counts):
-        data = {}
-
-        select = {"$match": self._build_mongo_query({"data_url": {"$ne": None}})}
-
-        if counts:
-            project = {"$project": {"_id": 0, "value": "$_id", "count": "$count"}}
-        else:
-            project = {"$project": {"_id": 0, "value": "$_id"}}
-
-        if metadata:
-            data["metadata"] = {}
-            for metadata_key in metadata:
-                group = {
-                    "$group": {"_id": f"$metadata.{metadata_key}", "count": {"$sum": 1}}
-                }
-                data["metadata"][f"{metadata_key}"] = list(
-                    self.collection.aggregate([select, group, project])
-                )
-
-        if structure_families:
-            group = {"$group": {"_id": "$structure_family", "count": {"$sum": 1}}}
-            data["structure_families"] = list(
-                self.collection.aggregate([select, group, project])
-            )
-
-        if specs:
-            group = {"$group": {"_id": "$specs", "count": {"$sum": 1}}}
-            distinct_list = list(self.collection.aggregate([select, group, project]))
-            data["specs"] = distinct_list
-
-        return data
-
-    def sort(self, sorting):
-        return self.new_variation(sorting=sorting)
-
-    def keys(self):
-        return KeysView(lambda: len(self), self._keys_slice)
-
-    def values(self):
-        return ValuesView(lambda: len(self), self._items_slice)
-
-    def items(self):
-        return ItemsView(lambda: len(self), self._items_slice)
-
-    def _keys_slice(self, start, stop, direction):
-        assert direction == 1, "direction=-1 should be handled by the client"
-        skip = start or 0
-        if stop is not None:
-            limit = stop - skip
-        else:
-            limit = None
-        for doc in self.collection.find(
-            self._build_mongo_query({"data_url": {"$ne": None}}),
-            skip=skip,
-            limit=limit,
-        ):
-            yield doc["key"]
-
-    def _items_slice(self, start, stop, direction):
-        assert direction == 1, "direction=-1 should be handled by the client"
-        skip = start or 0
-        if stop is not None:
-            limit = stop - skip
-        else:
-            limit = None
-
-        for doc in self.collection.find(
-            self._build_mongo_query({"data_url": {"$ne": None}}),
-            skip=skip,
-            limit=limit,
-        ):
-            class_ = _adapter_class_by_family[StructureFamily(doc["structure_family"])]
-            yield doc["key"], class_(self.database, doc["key"])
-
-    def apply_mongo_query(self, query):
-        return self.new_variation(
-            queries=self.queries + [query],
-        )
-
-
-def contains(query, catalog):
-    # In MongoDB, checking that an item is in an array looks
-    # just like equality.
-    # https://www.mongodb.com/docs/manual/tutorial/query-arrays/
-    return catalog.apply_mongo_query({f"metadata.{query.key}": query.value})
-
-
-def comparison(query, catalog):
-    OPERATORS = {
-        Operator.lt: "$lt",
-        Operator.le: "$lte",
-        Operator.gt: "$gt",
-        Operator.ge: "$gte",
-    }
-    return catalog.apply_mongo_query(
-        {f"metadata.{query.key}": {OPERATORS[query.operator]: query.value}}
-    )
-
-
-def eq(query, catalog):
-    return catalog.apply_mongo_query({f"metadata.{query.key}": query.value})
-
-
-def noteq(query, catalog):
-    return catalog.apply_mongo_query({f"metadata.{query.key}": {"$ne": query.value}})
-
-
-def regex(query, catalog):
-    options = "" if query.case_sensitive else "i"
-    return catalog.apply_mongo_query(
-        {f"metadata.{query.key}": {"$regex": query.pattern, "$options": options}}
-    )
-
-
-def _in(query, catalog):
-    if not isinstance(query.value, list):
-        query.value = [query.value]
-    return catalog.apply_mongo_query({f"metadata.{query.key}": {"$in": query.value}})
-
-
-def notin(query, catalog):
-    if not isinstance(query.value, list):
-        query.value = [query.value]
-    return catalog.apply_mongo_query({f"metadata.{query.key}": {"$nin": query.value}})
-
-
-def full_text_search(query, catalog):
-    # First if this catalog is backed by mongomock, which does not support $text queries.
-    # Avoid importing mongomock if it is not already imported.
-    if "mongomock" in modules:
-        import mongomock
-
-        if isinstance(catalog.database.client, mongomock.MongoClient):
-            # Do the query in memory.
-            # For huge MongoAdapters this will be slow, but if you are attempting
-            # full text search on a large mongomock-backed MongoAdapter,
-            # you have made your choices! :-)
-            return MapAdapter(dict(catalog)).search(query)
-
-    return catalog.apply_mongo_query(
-        {"$text": {"$search": query.text, "$caseSensitive": query.case_sensitive}},
-    )
-
-
-def keys_filter(query, catalog):
-    return catalog.apply_mongo_query({"key": {"$in": list(query.keys)}})
-
-
-MongoAdapter.register_query(Contains, contains)
-MongoAdapter.register_query(Comparison, comparison)
-MongoAdapter.register_query(Eq, eq)
-MongoAdapter.register_query(NotEq, noteq)
-MongoAdapter.register_query(Regex, regex)
-MongoAdapter.register_query(In, _in)
-MongoAdapter.register_query(KeysFilter, keys_filter)
-MongoAdapter.register_query(NotIn, notin)
-MongoAdapter.register_query(FullText, full_text_search)
-
-
-def safe_path(path):
-    if platform == "win32" and path[0] == "/":
-        path = path[1:]
-    return Path(path)
-
-
-_adapter_class_by_family = {
-    StructureFamily.array: WritingArrayAdapter,
-    StructureFamily.dataframe: WritingDataFrameAdapter,
-    StructureFamily.sparse: WritingCOOAdapter,
-}
diff --git a/databroker/mongo_normalized.py b/databroker/mongo_normalized.py
index dd55b3b0f..cf111ceea 100644
--- a/databroker/mongo_normalized.py
+++ b/databroker/mongo_normalized.py
@@ -1432,10 +1432,14 @@ def _build_event_stream(self, *, run_start_uid, stream_name, is_complete):
                 },
             ]
         )
-        (result,) = cursor
-        cutoff_seq_num = (
-            1 + result["highest_seq_num"]
-        )  # `1 +` because we use a half-open interval
+        results = list(cursor)
+        if results:
+            (result,) = results
+            cutoff_seq_num = (
+                1 + result["highest_seq_num"]
+            )  # `1 +` because we use a half-open interval
+        else:
+            cutoff_seq_num = 1
         object_names = event_descriptors[0]["object_keys"]
         run = self[run_start_uid]
         mapping = OneShotCachedMap(
diff --git a/databroker/tests/test_broker.py b/databroker/tests/test_broker.py
index f239bb3bc..7f691cd99 100644
--- a/databroker/tests/test_broker.py
+++ b/databroker/tests/test_broker.py
@@ -4,6 +4,7 @@
 import tempfile
 import os
 import logging
+import packaging
 import sys
 import string
 import time as ttime
@@ -24,14 +25,14 @@
 from databroker._core import DOCT_NAMES
 from databroker.tests.utils import get_uids
 
-if sys.version_info >= (3, 5):
-    from bluesky.plans import count
-    from bluesky.plan_stubs import trigger_and_read, configure
-    from bluesky.preprocessors import (monitor_during_wrapper,
-                                       run_decorator,
-                                       baseline_wrapper,
-                                       stage_wrapper,
-                                       pchain)
+from bluesky import __version__ as bluesky_version
+from bluesky.plans import count
+from bluesky.plan_stubs import trigger_and_read, configure, one_shot
+from bluesky.preprocessors import (monitor_during_wrapper,
+                                   run_decorator,
+                                   baseline_wrapper,
+                                   stage_wrapper,
+                                   pchain)
 
 logger = logging.getLogger(__name__)
 
@@ -86,11 +87,22 @@ def test_uid_list_multiple_headers(db, RE, hw):
 
 def test_no_descriptors(db, RE):
     RE.subscribe(db.insert)
-    uid, = get_uids(RE(count([])))
+    # Specifying per_shot as 'not None' turns off pre-declaring the stream and
+    # results in no descriptors.
+    uid, = get_uids(RE(count([], per_shot=one_shot), print))
     header = db[uid]
     assert [] == header.descriptors
 
 
+def test_no_events(db, RE):
+    if packaging.version.parse(bluesky_version) < packaging.version.parse("1.11.0"):
+        pytest.skip("This test relies on the pre-declare streams feature added in bluesky 1.11")
+    RE.subscribe(db.insert)
+    uid, = get_uids(RE(count([])))
+    header = db[uid]
+    header.table()
+
+
 def test_get_events(db, RE, hw):
     RE.subscribe(db.insert)
     uid, = get_uids(RE(count([hw.det])))
@@ -691,6 +703,7 @@ def test_dict_header(db, RE, hw):
         h['events']
 
 
+@pytest.mark.xfail(reason="Possibly broken by predeclare, need investigation")
 def test_config_data(db, RE, hw):
     # simple case: one Event Descriptor, one stream
     RE.subscribe(db.insert)
diff --git a/databroker/tests/test_experimental.py b/databroker/tests/test_experimental.py
deleted file mode 100644
index ac5baa5bb..000000000
--- a/databroker/tests/test_experimental.py
+++ /dev/null
@@ -1,558 +0,0 @@
-import contextlib
-import string
-
-import dask.array
-import dask.dataframe
-import httpx
-import numpy
-import pandas
-import pytest
-import sparse
-from tiled.client import Context, from_context
-from tiled.queries import (
-    Contains,
-    Comparison,
-    Eq,
-    FullText,
-    In,
-    Key,
-    NotEq,
-    NotIn,
-    Regex,
-)
-from tiled.server.app import build_app, build_app_from_config
-from tiled.structures.core import Spec
-from tiled.structures.sparse import COOStructure
-from tiled.validation_registration import ValidationRegistry
-
-from ..experimental.server_ext import MongoAdapter
-from ..experimental.schemas import DocumentRevision
-
-# from .test_access_policy import enter_password
-
-
-API_KEY = "secret"
-validation_registry = ValidationRegistry()
-validation_registry.register("SomeSpec", lambda *args, **kwargs: None)
-validation_registry.register("AnotherSpec", lambda *args, **kwargs: None)
-validation_registry.register("AnotherOtherSpec", lambda *args, **kwargs: None)
-
-
-@pytest.fixture
-def client(tmpdir):
-    tree = MongoAdapter.from_mongomock(tmpdir)
-    app = build_app(tree, validation_registry=validation_registry)
-    with Context.from_app(app) as context:
-        client = from_context(context)
-        yield client
-
-
-def test_write_array(client):
-    test_array = numpy.ones((5, 7))
-
-    metadata = {"scan_id": 1, "method": "A"}
-    specs = [Spec("SomeSpec")]
-    references = [{"label": "test", "url": "http://www.test.com"}]
-
-    node = client.write_array(
-        test_array, metadata=metadata, specs=specs, references=references
-    )
-
-    results = client.search(Key("scan_id") == 1)
-    result = results.values().first()
-    result_array = result.read()
-
-    numpy.testing.assert_equal(result_array, test_array)
-    assert result.metadata == node.metadata == metadata
-    assert result.specs == node.specs == specs
-    assert result.references == node.references == references
-
-
-def test_write_dataframe(client):
-    dummy_array = numpy.ones((5, 7))
-
-    data = {
-        "Column1": dummy_array[0],
-        "Column2": dummy_array[1],
-        "Column3": dummy_array[2],
-        "Column4": dummy_array[3],
-        "Column5": dummy_array[4],
-    }
-
-    test_dataframe = pandas.DataFrame(data)
-    metadata = {"scan_id": 1, "method": "A"}
-    specs = [Spec("SomeSpec")]
-    references = [{"label": "test", "url": "http://www.test.com"}]
-
-    node = client.write_dataframe(
-        test_dataframe, metadata=metadata, specs=specs, references=references
-    )
-
-    results = client.search(Key("scan_id") == 1)
-    result = results.values().first()
-    result_dataframe = result.read()
-
-    pandas.testing.assert_frame_equal(result_dataframe, test_dataframe)
-    # slicing into DataFrameClient returns ArrayClient
-    result_array = result["Column1"][:]
-    assert numpy.array_equal(result_array, dummy_array[0])
-    assert result.metadata == node.metadata == metadata
-    assert result.specs == node.specs == specs
-    assert result.references == node.references == references
-
-
-def test_queries(client):
-    keys = list(string.ascii_lowercase)
-
-    for letter, number in zip(keys, range(26)):
-        metadata = {"letter": letter, "number": number}
-        array = number * numpy.ones(10)
-
-        client.write_array(array, metadata=metadata)
-
-    test1 = client.search(Eq("letter", "a"))
-    numpy.testing.assert_equal(
-        test1.values()[0].read(), test1.values()[0].metadata["number"] * numpy.ones(10)
-    )
-    test2 = client.search(Contains("number", 1))
-    numpy.testing.assert_equal(
-        test2.values()[0].read(), test2.values()[0].metadata["number"] * numpy.ones(10)
-    )
-    test3 = client.search(Comparison("gt", "number", 24))
-    numpy.testing.assert_equal(
-        test3.values()[0].read(), test3.values()[0].metadata["number"] * numpy.ones(10)
-    )
-    test4 = client.search(FullText("y"))
-    numpy.testing.assert_equal(
-        test4.values()[0].read(), test4.values()[0].metadata["number"] * numpy.ones(10)
-    )
-    test5 = client.search(Regex("letter", "^c$"))
-    numpy.testing.assert_equal(
-        test5.values()[0].read(), test5.values()[0].metadata["number"] * numpy.ones(10)
-    )
-    test6 = client.search(NotEq("letter", "a"))
-    # The first result should not be "a"
-    assert test6.values()[0].metadata["letter"] != "a"
-
-    test7 = client.search(In("letter", ["a", "b"]))
-    numpy.testing.assert_equal(
-        test7.values()[0].read(), test7.values()[0].metadata["number"] * numpy.ones(10)
-    )
-    numpy.testing.assert_equal(
-        test7.values()[1].read(), test7.values()[1].metadata["number"] * numpy.ones(10)
-    )
-
-    test8 = client.search(NotIn("letter", ["a"]))
-    # The first result should not be "a"
-    assert test8.values()[0].metadata["letter"] != "a"
-
-
-def test_delete(client):
-    # For dataframes
-    dummy_array = numpy.ones((5, 5))
-
-    data = {
-        "Column1": dummy_array[0],
-        "Column2": dummy_array[1],
-        "Column3": dummy_array[2],
-        "Column4": dummy_array[3],
-        "Column5": dummy_array[4],
-    }
-
-    test_dataframe = pandas.DataFrame(data)
-
-    x = client.write_dataframe(
-        test_dataframe,
-        metadata={"scan_id": 1, "method": "A"},
-        specs=["SomeSpec"],
-        references=[{"label": "test", "url": "http://www.test.com"}],
-    )
-
-    del client[x.item["id"]]
-
-    assert x.item["id"] not in client
-
-    # For arrays
-    test_array = numpy.ones((5, 5))
-
-    y = client.write_array(
-        test_array,
-        metadata={"scan_id": 1, "method": "A"},
-        specs=["SomeSpec"],
-        references=[{"label": "test", "url": "http://www.test.com"}],
-    )
-
-    del client[y.item["id"]]
-
-    assert y.item["id"] not in client
-
-
-def test_write_array_chunked(client):
-    a = dask.array.arange(24).reshape((4, 6)).rechunk((2, 3))
-
-    metadata = {"scan_id": 1, "method": "A"}
-    specs = [Spec("SomeSpec")]
-    references = [{"label": "test", "url": "http://www.test.com"}]
-    client.write_array(a, metadata=metadata, specs=specs, references=references)
-
-    results = client.search(Key("scan_id") == 1)
-    result = results.values().first()
-    result_array = result.read()
-
-    numpy.testing.assert_equal(result_array, a.compute())
-    assert result.metadata == metadata
-    assert result.specs == specs
-    assert result.references == references
-
-
-def test_write_dataframe_partitioned(client):
-    data = {f"Column{i}": (1 + i) * numpy.ones(10) for i in range(5)}
-    df = pandas.DataFrame(data)
-    ddf = dask.dataframe.from_pandas(df, npartitions=3)
-    metadata = {"scan_id": 1, "method": "A"}
-    specs = [Spec("SomeSpec")]
-    references = [{"label": "test", "url": "http://www.test.com"}]
-
-    client.write_dataframe(ddf, metadata=metadata, specs=specs, references=references)
-
-    results = client.search(Key("scan_id") == 1)
-    result = results.values().first()
-    result_dataframe = result.read()
-
-    pandas.testing.assert_frame_equal(result_dataframe, df)
-    assert result.metadata == metadata
-    # TODO In the future this will be accessible via result.specs.
-    assert result.specs == specs
-    assert result.references == references
-
-
-def test_write_sparse_full(client):
-    coo = sparse.COO(coords=[[0, 1], [2, 3]], data=[3.8, 4.0], shape=(4, 4))
-
-    metadata = {"scan_id": 1, "method": "A"}
-    specs = [Spec("SomeSpec")]
-    references = [{"label": "test", "url": "http://www.test.com"}]
-    client.write_sparse(
-        coords=coo.coords,
-        data=coo.data,
-        shape=coo.shape,
-        metadata=metadata,
-        specs=specs,
-        references=references,
-    )
-
-    results = client.search(Key("scan_id") == 1)
-    result = results.values().first()
-    result_array = result.read()
-
-    numpy.testing.assert_equal(result_array.todense(), coo.todense())
-    assert result.metadata == metadata
-    assert result.specs == specs
-    assert result.references == references
-
-
-def test_write_sparse_chunked(client):
-    metadata = {"scan_id": 1, "method": "A"}
-    specs = [Spec("SomeSpec")]
-    references = [{"label": "test", "url": "http://www.test.com"}]
-    N = 5
-    x = client.new(
-        "sparse",
-        COOStructure(shape=(2 * N,), chunks=((N, N),)),
-        metadata=metadata,
-        specs=specs,
-        references=references,
-    )
-    x.write_block(coords=[[2, 4]], data=[3.1, 2.8], block=(0,))
-    x.write_block(coords=[[0, 1]], data=[6.7, 1.2], block=(1,))
-
-    results = client.search(Key("scan_id") == 1)
-    result = results.values().first()
-    result_array = result.read()
-    assert numpy.array_equal(
-        result_array.todense(),
-        sparse.COO(
-            coords=[[2, 4, N + 0, N + 1]], data=[3.1, 2.8, 6.7, 1.2], shape=(10,)
-        ).todense(),
-    )
-
-    # numpy.testing.assert_equal(result_array, sparse.COO(coords=[0, 1, ]))
-    assert result.metadata == metadata
-    assert result.specs == specs
-    assert result.references == references
-
-
-def test_update_array_metadata(client):
-    # Update metadata in array client
-    test_array = numpy.ones((5, 5))
-
-    x = client.write_array(
-        test_array, metadata={"scan_id": 1, "method": "A"}, specs=["SomeSpec"]
-    )
-
-    new_arr_metadata = {"scan_id": 2, "method": "A"}
-    new_spec = [Spec("AnotherSpec")]
-    references = [{"label": "test", "url": "http://www.test.com"}]
-    x.update_metadata(new_arr_metadata, new_spec, references)
-
-    # validate local data after update request
-    assert x.metadata == new_arr_metadata
-    assert x.specs == new_spec
-    assert x.references == references
-
-    # Update metadata again to create another entry in revisions
-    newer_arr_metadata = {"scan_id": 2, "method": "B"}
-    newer_spec = [Spec("AnotherOtherSpec")]
-    new_references = [{"label": "updated_test", "url": "http://www.updatedtest.com"}]
-    x.update_metadata(newer_arr_metadata, newer_spec, new_references)
-
-    # Increase the size of revisions for additonal testing
-    latest_arr_metadata = {"scan_id": 2, "method": "C"}
-    x.update_metadata(latest_arr_metadata)
-
-    results = client.search(Key("scan_id") == 2)
-    result = results.values().first()
-
-    # validate remote data after update request
-    assert result.metadata == latest_arr_metadata
-    assert result.specs == newer_spec
-    assert result.references == new_references
-
-    rev_document = {
-        "key": result.item["id"],
-        "revision": result.metadata_revisions[0]["revision"],
-    }
-    rev_document.update(result.metadata_revisions[0]["attributes"])
-    assert DocumentRevision.from_json(rev_document)
-
-    assert len(result.metadata_revisions[0:2]) == 2
-    assert len(result.metadata_revisions) == len(result.metadata_revisions[:])
-
-    result.metadata_revisions.delete_revision(0)
-    assert len(result.metadata_revisions[:]) == 2
-
-
-def test_update_dataframe_metadata(client):
-    test_array = numpy.ones((5, 5))
-
-    # Update metadata in dataframe client
-    data = {
-        "Column1": test_array[0],
-        "Column2": test_array[1],
-        "Column3": test_array[2],
-        "Column4": test_array[3],
-        "Column5": test_array[4],
-    }
-
-    test_dataframe = pandas.DataFrame(data)
-
-    y = client.write_dataframe(
-        test_dataframe, metadata={"scan_id": 1, "method": "A"}, specs=["SomeSpec"]
-    )
-
-    new_df_metadata = {"scan_id": 2, "method": "A"}
-    new_spec = [Spec("AnotherSpec")]
-    references = [{"label": "test", "url": "http://www.test.com"}]
-    y.update_metadata(new_df_metadata, new_spec, references)
-
-    # validate local data after update request
-    assert y.metadata == new_df_metadata
-    assert y.specs == new_spec
-    assert y.references == references
-
-    # Update metadata again to create another entry in revisions
-    newer_df_metadata = {"scan_id": 2, "method": "B"}
-    newer_spec = [Spec("AnotherOtherSpec")]
-    new_references = [{"label": "updated_test", "url": "http://www.updatedtest.com"}]
-    y.update_metadata(newer_df_metadata, newer_spec, new_references)
-
-    # Increase the size of revisions for additonal testing
-    latest_arr_metadata = {"scan_id": 2, "method": "C"}
-    y.update_metadata(latest_arr_metadata)
-
-    results = client.search(Key("scan_id") == 2)
-    result = results.values().first()
-
-    # validate remote data after update request
-    assert result.metadata == latest_arr_metadata
-    assert result.specs == newer_spec
-    assert result.references == new_references
-
-    rev_document = {
-        "key": result.item["id"],
-        "revision": result.metadata_revisions[0]["revision"],
-    }
-    rev_document.update(result.metadata_revisions[0]["attributes"])
-    assert DocumentRevision.from_json(rev_document)
-
-    assert len(result.metadata_revisions[0:2]) == 2
-    assert len(result.metadata_revisions) == len(result.metadata_revisions[:])
-
-    result.metadata_revisions.delete_revision(0)
-    assert len(result.metadata_revisions[:]) == 2
-
-
-@contextlib.contextmanager
-def fail_with_status_code(status_code):
-    with pytest.raises(httpx.HTTPStatusError) as info:
-        yield
-    assert info.value.response.status_code == status_code
-
-
-def test_simple_access_policy(tmpdir, enter_password):
-    config = {
-        "authentication": {
-            "providers": [
-                {
-                    "provider": "toy",
-                    "authenticator": "tiled.authenticators:DictionaryAuthenticator",
-                    "args": {
-                        "users_to_passwords": {
-                            "alice": "secret1",
-                            "bob": "secret2",
-                            "cara": "secret3",
-                        }
-                    },
-                }
-            ],
-        },
-        "trees": [
-            {
-                "path": "/",
-                "tree": "databroker.experimental.server_ext:MongoAdapter.from_mongomock",
-                "args": {"directory": tmpdir},
-                "access_control": {
-                    "access_policy": "tiled.access_policies:SimpleAccessPolicy",
-                    "args": {
-                        "provider": "toy",
-                        "access_lists": {
-                            "alice": [],
-                            "cara": "tiled.access_policies:ALL_ACCESS",
-                        },
-                    },
-                },
-            }
-        ],
-    }
-    with Context.from_app(build_app_from_config(config), token_cache=tmpdir) as context:
-        # User with all access
-        with enter_password("secret3"):
-            client = from_context(
-                context, username="cara", prompt_for_reauthentication=True
-            )
-        client.write_array([1, 2, 3])
-        assert len(list(client)) == 1
-        # User with no access
-        with enter_password("secret1"):
-            client = from_context(
-                context, username="alice", prompt_for_reauthentication=True
-            )
-        assert len(list(client)) == 0
-        # User with implicitly no access (not mentioned in policy)
-        with enter_password("secret2"):
-            client = from_context(
-                context, username="bob", prompt_for_reauthentication=True
-            )
-        assert len(list(client)) == 0
-
-
-def test_distinct(client):
-    # Begin of test data generation
-
-    # Added additional field in metadata to implement consecutive search and distinct queries
-    for i in range(10):
-        if i < 5:
-            group = "A"
-        else:
-            group = "B"
-
-        if i % 2 == 0:
-            subgroup = "even"
-            specs = [Spec("test")]
-        else:
-            subgroup = "odd"
-            specs = [Spec("test"), Spec("moreTest")]
-
-        if i == 0:
-            tag = "Zero"
-        else:
-            for j in range(2, int(i / 2) + 1):
-                if (i % j) == 0:
-                    tag = "NotPrime"
-                    break
-            else:
-                tag = "Prime"
-
-        df = pandas.DataFrame({"a": i * numpy.ones(10)})
-        metadata = {"group": group, "subgroup": subgroup, "tag": tag}
-
-        client.write_dataframe(df, metadata=metadata, specs=specs)
-
-    # End of test data generation
-
-    # Not Counting
-    results = client.search(Key("group") == "B").distinct(
-        "tag", structure_families=True, specs=True, counts=False
-    )
-
-    # Results are retrieved from the database as an unsorted list.
-    # They are sorted by count to validate them during the test run.
-    results["metadata"]["tag"].sort(key=lambda k: k["value"])
-    results["specs"].sort(key=lambda k: k["value"])
-
-    expected = {
-        "metadata": {
-            "tag": [
-                {"value": "NotPrime", "count": None},
-                {"value": "Prime", "count": None},
-            ]
-        },
-        "specs": [
-            {"value": [{"name": "test", "version": None}], "count": None},
-            {
-                "value": [
-                    {"name": "test", "version": None},
-                    {"name": "moreTest", "version": None},
-                ],
-                "count": None,
-            },
-        ],
-        "structure_families": [{"value": "dataframe", "count": None}],
-    }
-
-    assert results["metadata"] == expected["metadata"]
-    assert results["specs"] == expected["specs"]
-    assert results["structure_families"] == expected["structure_families"]
-
-    # Counting
-    results = client.search(Key("group") == "B").distinct(
-        "tag", structure_families=True, specs=True, counts=True
-    )
-
-    # Results are retrieved from the database as an unsorted list.
-    # They are sorted by count to validate them during the test run.
-    results["metadata"]["tag"].sort(key=lambda k: k["count"])
-    results["specs"].sort(key=lambda k: k["count"])
-
-    expected = {
-        "metadata": {
-            "tag": [{"value": "Prime", "count": 2}, {"value": "NotPrime", "count": 3}]
-        },
-        "specs": [
-            {"value": [{"name": "test", "version": None}], "count": 2},
-            {
-                "value": [
-                    {"name": "test", "version": None},
-                    {"name": "moreTest", "version": None},
-                ],
-                "count": 3,
-            },
-        ],
-        "structure_families": [
-            {"value": "dataframe", "count": 5},
-        ],
-    }
-
-    assert results["metadata"] == expected["metadata"]
-    assert results["specs"] == expected["specs"]
-    assert results["structure_families"] == expected["structure_families"]