Repository: square/bionic

Issue:
The "provider" concept is too broad
Currently Bionic's internals use a concept of a "provider" object which contains the logic for computing each entity value. This class (or family of classes) handles two categories of things:

1. Generate the task objects responsible for applying a particular data transformation. Each provider applies exactly one transformation. (That transformation may be composed of several others, but it's consistent for a given provider). 
2. Serve as the source of truth for entity properties (names, docstrings, protocols, etc.). A provider is usually responsible for one entity, but can sometimes have more.

The fact that a provider is not 1:1 with an entity suggests that we should separate these two concepts. In fact, as we add non-entity descriptors, we'll soon want to apply transformations that aren't associated with any entity at all. (For example, constructing or deconstructing a tuple.)

Furthermore, providers are structured in such a way that they can be incrementally transformed by decorator functions. This means that during the decoration process, a provider can have an incomplete or invalid configuration, with the expectation that it will be cleaned up or rejected by the final decorator prior to being added to a Flow. This need to support both clean and unclean states makes the provider interface and data model even more complicated.

To address this two issues, I propose splitting out two additional classes

1. An `EntityDefinition` class which acts as the source of truth for entity properties. This object is immutable and always in a clean, valid state.
2. A `DecorationAccumulator` class which holds the messy entity configuration data built up by Bionic decorators. When a decorated function is incorporated into a flow, this accumulator will be validated, normalized, and converted into a provider and one or more `EntityDefinition`s.

The provider concept will remain, but it will only be responsible for computing values by applying transformations.

Discussion:

Patch:
diff --git a/bionic/datatypes.py b/bionic/datatypes.py
index 2191390..9607976 100644
--- a/bionic/datatypes.py
+++ b/bionic/datatypes.py
@@ -7,6 +7,22 @@ import attr
 from .util import ImmutableSequence, ImmutableMapping
 
 
+@attr.s(frozen=True)
+class EntityDefinition:
+    """
+    Describes the immutable properties of an entity. These properties generally have
+    to do with the entity's "contract": the assumptions other parts of the system can
+    make about its value. However, this does *not* include the way the entity's value
+    is determined; this is configured separately and can be changed more easily.
+    """
+
+    name = attr.ib()
+    protocol = attr.ib()
+    doc = attr.ib()
+    can_persist = attr.ib(default=True)
+    can_memoize = attr.ib(default=True)
+
+
 @attr.s(frozen=True)
 class TaskKey:
     """
diff --git a/bionic/decoration.py b/bionic/decoration.py
index ab2aa97..e6fa8e4 100644
--- a/bionic/decoration.py
+++ b/bionic/decoration.py
@@ -16,28 +16,57 @@ this accumulator object and uses it to define a new entity.
 """
 
 import attr
+import warnings
 
+from .exception import AttributeValidationError
 from .provider import FunctionProvider
+from .util import oneline
 
 
 @attr.s
 class DecorationAccumulator:
     provider = attr.ib()
 
+    protocol = attr.ib(default=None)
+    docs = attr.ib(default=None)
+    can_persist = attr.ib(default=None)
+    can_memoize = attr.ib(default=None)
+
     def wrap_provider(self, wrapper_fn, *args, **kwargs):
         self.provider = wrapper_fn(self.provider, *args, **kwargs)
 
-
-def decorator_wrapping_provider(wrapper_fn, *args, **kwargs):
+    def update_attr(
+        self, attr_name, attr_value, decorator_name, raise_if_already_set=True
+    ):
+        old_attr_value = getattr(self, attr_name)
+        if old_attr_value is not None:
+            message = f"""
+            Tried to use {decorator_name} with value {attr_value!r},
+            but this decorator was already used with value {old_attr_value!r}
+            """
+            if raise_if_already_set:
+                raise AttributeValidationError(oneline(message))
+            else:
+                preamble = """
+                Applying this type of decorator multiple times is deprecated and will
+                become an error condition in a future release; please remove all but
+                the uppermost uses of this decorator. Details:
+                """
+                warnings.warn(oneline(preamble) + "\n" + oneline(message))
+        setattr(self, attr_name, attr_value)
+
+
+def decorator_updating_accumulator(acc_update_func):
     """
-    Returns a decorator which transforms the accumulated provider on a decorated
-    function by applying the given wrapping function and arguments.
+    Creates a decorator which applies a transformation to the DecorationAccumulator
+    attached to the decorated function. (If no accumulator is attached, the decorator
+    will initialize one.)
     """
 
     def decorator(func):
         init_accumulator_if_not_set_on_func(func)
         acc = get_accumulator_from_func(func)
-        acc.wrap_provider(wrapper_fn, *args, **kwargs)
+        acc_update_func(acc)
         return func
 
     return decorator
diff --git a/bionic/decorators.py b/bionic/decorators.py
index b4b2030..6496c33 100644
--- a/bionic/decorators.py
+++ b/bionic/decorators.py
@@ -11,7 +11,8 @@ These are the decorators we expose to Bionic users.  They are used as follows:
 """
 
 from .datatypes import CodeVersion
-from .decoration import decorator_wrapping_provider
+from .decoration import decorator_updating_accumulator
+from .exception import AttributeValidationError
 from .provider import (
     GatherProvider,
     AttrUpdateProvider,
@@ -19,6 +20,7 @@ from .provider import (
     RenamingProvider,
     NameSplittingProvider,
 )
+from .util import oneline
 from . import interpret
 
 
@@ -59,8 +61,10 @@ def version(major=None, minor=None):
         A decorator which can be applied to an entity function.
     """
 
-    return decorator_wrapping_provider(
-        AttrUpdateProvider, "code_version", CodeVersion(major, minor)
+    return decorator_updating_accumulator(
+        lambda acc: acc.wrap_provider(
+            AttrUpdateProvider, "code_version", CodeVersion(major, minor)
+        )
     )
 
 
@@ -87,7 +91,9 @@ def persist(enabled):
     if not isinstance(enabled, bool):
         raise ValueError(f"Argument must be a boolean; got {enabled!r}")
 
-    return decorator_wrapping_provider(AttrUpdateProvider, "_can_persist", enabled)
+    return decorator_updating_accumulator(
+        lambda acc: acc.update_attr("can_persist", enabled, "@persist")
+    )
 
 
 def memoize(enabled):
@@ -109,7 +115,9 @@ def memoize(enabled):
     if not isinstance(enabled, bool):
         raise ValueError(f"Argument must be a boolean; got {enabled!r}")
 
-    return decorator_wrapping_provider(AttrUpdateProvider, "_can_memoize", enabled)
+    return decorator_updating_accumulator(
+        lambda acc: acc.update_attr("can_memoize", enabled, "@memoize")
+    )
 
 
 def changes_per_run(enabled=None):
@@ -189,7 +197,9 @@ def changes_per_run(enabled=None):
     if not isinstance(enabled, bool):
         raise ValueError(f"Argument must be a boolean; got {enabled!r}")
 
-    return decorator_wrapping_provider(AttrUpdateProvider, "changes_per_run", enabled)
+    return decorator_updating_accumulator(
+        lambda acc: acc.wrap_provider(AttrUpdateProvider, "changes_per_run", enabled)
+    )
 
 
 def output(name):
@@ -211,7 +221,9 @@ def output(name):
         A decorator which can be applied to an entity function.
     """
 
-    return decorator_wrapping_provider(RenamingProvider, name)
+    return decorator_updating_accumulator(
+        lambda acc: acc.wrap_provider(RenamingProvider, name)
+    )
 
 
 def outputs(*names):
@@ -239,7 +251,9 @@ def outputs(*names):
         A decorator which can be applied to an entity function.
     """
 
-    return decorator_wrapping_provider(NameSplittingProvider, names)
+    return decorator_updating_accumulator(
+        lambda acc: acc.wrap_provider(NameSplittingProvider, names)
+    )
 
 
 def docs(*docs):
@@ -262,12 +276,9 @@ def docs(*docs):
         A decorator which can be applied to an entity function.
     """
 
-    return decorator_wrapping_provider(AttrUpdateProvider, "docs", docs)
-
-
-# TODO I'd like to put a @protocols decorator here that exposes the
-# MultiProtocolUpdateProvider class, but that would collide with the
-# protocols.py module.  Let's do this in a later PR.
+    return decorator_updating_accumulator(
+        lambda acc: acc.update_attr("docs", docs, "@docs")
+    )
 
 
 def gather(over, also=None, into="gather_df"):
@@ -329,8 +340,14 @@ def gather(over, also=None, into="gather_df"):
     """
     over = interpret.str_or_seq_as_list(over)
     also = interpret.str_or_seq_or_none_as_list(also)
-    return decorator_wrapping_provider(
-        GatherProvider, primary_names=over, secondary_names=also, gathered_dep_name=into
+
+    return decorator_updating_accumulator(
+        lambda acc: acc.wrap_provider(
+            GatherProvider,
+            primary_names=over,
+            secondary_names=also,
+            gathered_dep_name=into,
+        )
     )
 
 
@@ -366,7 +383,10 @@ def pyplot(name=None, savefig_kwargs=None):
 
     if name is None:
         name = "pyplot"
-    return decorator_wrapping_provider(PyplotProvider, name, savefig_kwargs)
+
+    return decorator_updating_accumulator(
+        lambda acc: acc.wrap_provider(PyplotProvider, name, savefig_kwargs)
+    )
 
 
 immediate = persist(False)
diff --git a/bionic/deriver.py b/bionic/deriver.py
index b88bc3d..3631840 100644
--- a/bionic/deriver.py
+++ b/bionic/deriver.py
@@ -116,9 +116,7 @@ class EntityDeriver:
                     entity_name=entity_name,
                     case_key=task_key.case_key,
                     task_ix=task_ix,
-                    doc=self._flow_state.get_provider(entity_name).doc_for_name(
-                        entity_name
-                    ),
+                    doc=self._flow_state.get_entity_def(entity_name).doc,
                 )
 
                 for dep_state in state.dep_states:
@@ -249,26 +247,38 @@ class EntityDeriver:
         dinfo = self._get_or_create_dinfo_for_dnode(dnode)
         task = dinfo.tasks_by_key[task_key]
 
+        entity_names = [task_key.dnode.to_entity_name() for task_key in task.keys]
+
         dep_states = [
             self._get_or_create_task_state_for_key(dep_key) for dep_key in task.dep_keys
         ]
         # All keys in this task should point to the same provider, so the set below
         # should have exactly one element.
         (provider,) = set(
-            self._flow_state.get_provider(task_key.dnode.to_entity_name())
-            for task_key in task.keys
+            self._flow_state.get_provider(entity_name) for entity_name in entity_names
         )
         # And all the task keys should have the same case key.
         (case_key,) = set(task_key.case_key for task_key in task.keys)
 
+        entity_defs_by_name = {
+            entity_name: self._flow_state.get_entity_def(entity_name)
+            for entity_name in entity_names
+        }
+
         task_state = TaskState(
-            task=task, dep_states=dep_states, provider=provider, case_key=case_key,
+            task=task,
+            dep_states=dep_states,
+            case_key=case_key,
+            provider=provider,
+            entity_defs_by_name=entity_defs_by_name,
         )
 
         # Check that the provider configuration is valid.
-        if provider.attrs.changes_per_run and not provider.attrs.should_memoize():
+        if provider.attrs.changes_per_run and not task_state.should_memoize:
+            # TODO This message should say something like:
+            #    "Entity with name ..." or "Entities with names ..."
             message = f"""
-            Entity with names {provider.attrs.names!r} uses @changes_per_run with
+            Entity with names {entity_names!r} uses @changes_per_run with
             @memoize(False), which is not allowed. @changes_per_run computes once in
             a flow instance and memoizes the value. Memoization cannot be disabled
             for this entity.
diff --git a/bionic/flow.py b/bionic/flow.py
index 7ceba47..79085fd 100644
--- a/bionic/flow.py
+++ b/bionic/flow.py
@@ -19,7 +19,7 @@ import pandas as pd
 # A bit annoying that we have to rename this when we import it.
 from . import protocols as protos
 from .cache import LocalStore, GcsCloudStore, PersistentCache
-from .datatypes import CaseKey, VersioningPolicy, ResultGroup
+from .datatypes import CaseKey, EntityDefinition, VersioningPolicy, ResultGroup
 from .exception import (
     UndefinedEntityError,
     AlreadyDefinedEntityError,
@@ -31,7 +31,6 @@ from .provider import (
     ValueProvider,
     multi_index_from_case_keys,
     AttrUpdateProvider,
-    ProtocolUpdateProvider,
 )
 from .deriver import EntityDeriver, TaskKeyLogger
 from .descriptors.parsing import entity_dnode_from_descriptor
@@ -72,7 +71,8 @@ class FlowState(pyrs.PClass):
     """
 
     providers_by_name = pyrs.field(initial=pyrs.pmap())
-    default_provider_names = pyrs.field(initial=pyrs.pset())
+    entity_defs_by_name = pyrs.field(initial=pyrs.pmap())
+    default_entity_names = pyrs.field(initial=pyrs.pset())
     last_added_case_key = pyrs.field(initial=None)
 
     # These are used to keep track of whether a flow state is safe to reload.
@@ -93,6 +93,27 @@ class FlowState(pyrs.PClass):
     def touch(self):
         return self.set(is_blessed=False)
 
+    def define_entity(self, entity_def):
+        if entity_def.name in self.entity_defs_by_name:
+            raise AlreadyDefinedEntityError(entity_def.name)
+
+        return self._set_entity_def(entity_def).touch()
+
+    def delete_entity_defs(self, names):
+        state = self
+
+        for name in names:
+            if name not in self.entity_defs_by_name:
+                continue
+            state = state._erase_entity_def(name).touch()
+
+        return state.touch()
+
+    def get_entity_def(self, name):
+        if name not in self.entity_defs_by_name:
+            raise UndefinedEntityError.for_name(name)
+        return self.entity_defs_by_name[name]
+
     def get_provider(self, name):
         if name not in self.providers_by_name:
             raise UndefinedEntityError.for_name(name)
@@ -101,11 +122,11 @@ class FlowState(pyrs.PClass):
     def has_provider(self, name):
         return name in self.providers_by_name
 
-    def create_provider(self, name, protocol, doc):
+    def create_provider(self, name):
         if name in self.providers_by_name:
             raise AlreadyDefinedEntityError.for_name(name)
 
-        provider = ValueProvider([name], [protocol], [doc])
+        provider = ValueProvider([name])
         return self._set_provider(provider).touch()
 
     def install_provider(self, provider):
@@ -192,30 +213,49 @@ class FlowState(pyrs.PClass):
         if names != provider.attrs.names:
             values = [values[provider.attrs.names.index(name)] for name in names]
 
-        provider = provider.add_case(case_key, values)
+        tokens = []
+        for name, value in zip(names, values):
+            protocol = self.get_entity_def(name).protocol
+            protocol.validate(value)
+            tokens.append(protocol.tokenize(value))
+        provider = provider.add_case(case_key, values, tokens)
 
         return self._set_provider(provider).touch()
 
     def clear_providers(self, names):
         state = self
 
-        # Remember the original provider attributes per entity.
-        attrs_by_entity_name = {}
-        for name in names:
-            if not state.has_provider(name):
-                continue
-            provider = self.get_provider(name)
-            attrs_by_entity_name[name] = provider.attrs
+        # Remember the original entity definitions.
+        # TODO Here we delete each entity definition and then re-create it. The only
+        # effect this has is to reset the can_persist and can_memoize attributes to
+        # their default values. This shouldn't change any actual behavior, since these
+        # attributes don't matter for ValueProviders. However, I'm keeping the
+        # deletion/creation code in order to keep the entity definition attributes 100%
+        # consistent with their legacy provider equivalents. Once the old provider code
+        # has been removed, we can decide whether to remove the deletion/creation of
+        # the entity definitions.
+        original_entity_defs = [
+            state.get_entity_def(name)
+            for name in names
+            if name in state.entity_defs_by_name
+        ]
 
         # Delete the providers (or fail if not possible).
         state = state.delete_providers(names)
+        state = state.delete_entity_defs(names)
 
         # Recreate an empty version of each provider.
-        for name, attrs in attrs_by_entity_name.items():
-            name_ix = attrs.names.index(name)
-            protocol = attrs.protocols[name_ix]
-            doc = attrs.docs[name_ix]
-            state = state.create_provider(name, protocol, doc)
+        for entity_def in original_entity_defs:
+            state = state.define_entity(
+                # As noted above, we only preserve some of the attributes from the
+                # original definition.
+                EntityDefinition(
+                    name=entity_def.name,
+                    protocol=entity_def.protocol,
+                    doc=entity_def.doc,
+                )
+            )
+            state = state.create_provider(entity_def.name)
 
         return state.touch()
 
@@ -246,30 +286,43 @@ class FlowState(pyrs.PClass):
 
         return state.touch()
 
-    def mark_all_providers_default(self):
-        state = self
+    def mark_all_entities_default(self):
+        new_default_names = pyrs.pset(self.entity_defs_by_name.keys())
+        return self.set(default_entity_names=new_default_names).touch()
 
-        new_default_names = pyrs.pset(state.providers_by_name.keys())
-        state = state.set(default_provider_names=new_default_names)
+    def _erase_entity_def(self, name):
+        assert name not in self.providers_by_name
 
-        return state.touch()
+        state = self
+        state = state.set(entity_defs_by_name=state.entity_defs_by_name.remove(name),)
+        if name in state.default_entity_names:
+            state = state.set(
+                default_entity_names=(state.default_entity_names.remove(name))
+            )
+        return state
 
     def _erase_provider(self, name):
         state = self
         if name in state.providers_by_name:
-            state = state.set(providers_by_name=state.providers_by_name.remove(name))
-        if name in state.default_provider_names:
-            state = state.set(
-                default_provider_names=(state.default_provider_names.remove(name))
-            )
+            state = state.set(providers_by_name=state.providers_by_name.remove(name),)
         return state
 
+    def _set_entity_def(self, entity_def):
+        return self.set(
+            entity_defs_by_name=self.entity_defs_by_name.set(
+                entity_def.name, entity_def
+            ),
+        )
+
     def _set_provider(self, provider):
+        for name in provider.attrs.names:
+            assert name in self.entity_defs_by_name
+
         state = self
         for name in provider.attrs.names:
             state = state._erase_provider(name)
             state = state.set(
-                providers_by_name=state.providers_by_name.set(name, provider)
+                providers_by_name=state.providers_by_name.set(name, provider),
             )
         return state
 
@@ -350,7 +403,9 @@ class FlowBuilder:
             )
             doc = docstring
 
-        self._state = self._state.create_provider(name, protocol, doc)
+        self._state = self._state.define_entity(
+            EntityDefinition(name=name, protocol=protocol, doc=doc)
+        ).create_provider(name)
 
     def assign(
         self, name, value=None, values=None, protocol=None, doc=None, docstring=None
@@ -392,12 +447,16 @@ class FlowBuilder:
             )
             doc = docstring
 
+        # TODO We can remove this validation, since it also happens in add_case below.
         for value in values:
             protocol.validate(value)
 
         state = self._state
 
-        state = state.create_provider(name, protocol, doc)
+        state = state.define_entity(
+            EntityDefinition(name=name, protocol=protocol, doc=doc)
+        )
+        state = state.create_provider(name)
         for value in values:
             case_key = CaseKey([(name, value, protocol.tokenize(value))])
             state = state.add_case(case_key, [name], [value])
@@ -430,14 +489,13 @@ class FlowBuilder:
         state = self._state
 
         state = state.clear_providers([name])
-        provider = state.get_provider(name)
-        # This provider must have a single name and single protocol; otherwise
-        # we wouldn't have been able to clear it.
-        (protocol,) = provider.attrs.protocols
 
-        protocol.validate(value)
+        protocol = state.get_entity_def(name).protocol
 
         for value in values:
+            # TODO We can remove this validation, since it also happens in add_case
+            # below.
+            protocol.validate(value)
             case_key = CaseKey([(name, value, protocol.tokenize(value))])
             state = state.add_case(case_key, [name], [value])
 
@@ -520,12 +578,13 @@ class FlowBuilder:
 
         names = [name for name, value in name_value_pairs]
         state = state.group_names_together(names)
-        provider = state.get_provider(names[0])
 
         values = []
         case_nvt_tuples = []
         for name, value in name_value_pairs:
-            protocol = provider.protocol_for_name(name)
+            protocol = state.get_entity_def(name).protocol
+            # TODO Both the validation and tokenization are also happening in
+            # state.add_case below; maybe we can remove some duplicate work.
             protocol.validate(value)
             token = protocol.tokenize(value)
 
@@ -574,7 +633,7 @@ class FlowBuilder:
             The entities to be deleted.
         """
 
-        self._state = self._state.delete_providers(names)
+        self._state = self._state.delete_providers(names).delete_entity_defs(names)
 
     def merge(self, flow, keep="error", allow_name_match=False):
         """
@@ -678,7 +737,7 @@ class FlowBuilder:
                 conflict.resolve("new", "conflicting definition is default")
                 continue
 
-            if conflict.old_protocol is conflict.new_protocol:
+            if conflict.old_entity_def.protocol is conflict.new_entity_def.protocol:
                 if conflict.new_is_only_declaration:
                     conflict.resolve(
                         "old",
@@ -776,6 +835,7 @@ class FlowBuilder:
         names_to_delete = [conflict.name for conflict in conflicts_keeping_new]
         try:
             cur_state = cur_state.delete_providers(names_to_delete)
+            cur_state = cur_state.delete_entity_defs(names_to_delete)
         except IncompatibleEntityError as e:
             raise IncompatibleEntityError("Merge failure: " + e)
 
@@ -795,6 +855,9 @@ class FlowBuilder:
                 and provider.attrs.orig_flow_name is None
             ):
                 provider = AttrUpdateProvider(provider, "orig_flow_name", new_flow_name)
+            for name in provider.attrs.names:
+                new_entity_def = new_state.get_entity_def(name)
+                cur_state = cur_state.define_entity(new_entity_def)
             cur_state = cur_state.install_provider(provider)
 
         self._state = cur_state
@@ -820,69 +883,81 @@ class FlowBuilder:
         """
 
         decoration.init_accumulator_if_not_set_on_func(func)
-        provider = decoration.pop_accumulator_from_func(func).provider
-
-        if provider.attrs.protocols is None:
-            provider = ProtocolUpdateProvider(provider, DEFAULT_PROTOCOL)
-        if provider.attrs.docs is None:
-            docs = [None] * len(provider.attrs.names)
-            provider = AttrUpdateProvider(provider, "docs", docs)
-        if provider.attrs._can_persist is None:
-            provider = AttrUpdateProvider(provider, "_can_persist", True)
-        if provider.attrs._can_memoize is None:
-            provider = AttrUpdateProvider(provider, "_can_memoize", True)
+        acc = decoration.pop_accumulator_from_func(func)
+
+        provider = acc.provider
+
+        # TODO Data like `changes_per_run` and `version` should probably also be stored
+        # in the DecorationAccumulator and then attached to some kind of
+        # FunctionAttributes object.
         if provider.attrs.changes_per_run is None:
             provider = AttrUpdateProvider(provider, "changes_per_run", False)
 
-        if not (provider.attrs.should_persist() or provider.attrs.should_memoize()):
-            raise ValueError(
-                oneline(
-                    f"""
-                Attempted to set both persist and memoize to False.
-                At least one form of storage must be enabled for entities:
-                {provider.attrs.names!r}"""
-                )
-            )
-        if len(provider.attrs.protocols) != len(provider.attrs.names):
-            names = provider.attrs.names
-            protocols = provider.attrs.protocols
-            raise ValueError(
-                oneline(
-                    f"""
-                Number of protocols must match the number of names;
-                got {len(names)} names {tuple(names)!r} and
-                {len(protocols)} protocols {tuple(protocols)!r}"""
-                )
-            )
-        if len(provider.attrs.docs) != len(provider.attrs.names):
-            names = provider.attrs.names
-            docs = provider.attrs.docs
-            message = oneline(
-                f"""
-                Number of docs must match the number of names;
+        names = provider.attrs.names
+
+        protocol = acc.protocol
+        if protocol is None:
+            protocol = DEFAULT_PROTOCOL
+
+        docs = acc.docs
+        orig_docstring = provider.get_source_func().__doc__
+        if docs is None:
+            if orig_docstring is not None and len(names) > 1:
+                preamble = """
+                Using a single doc string for a multi-output function is
+                deprecated and will become an error condition in a future
+                release; use the ``@docs`` decorator to specify
+                one doc string for each entity.  Details:
+                """
+                details = f"""
                 got {len(names)} names {tuple(names)!r} and
-                {len(docs)} docs {tuple(docs)!r}"""
-            )
-            if len(provider.attrs.docs) == 1:
-                prefix = oneline(
-                    """
-                    Using a single doc string for a multi-output function is
-                    deprecated and will become an error condition in a future
-                    release; use the ``@docs`` decorator to specify
-                    one doc string for each entity.  Details:"""
-                )
-                warnings.warn(prefix + "\n" + message)
+                just one docstring {orig_docstring}
+                """
+                warnings.warn(oneline(preamble) + "\n" + oneline(details))
 
-                docs = [docs[0]] * len(names)
-                provider = AttrUpdateProvider(
-                    provider, "docs", docs, allow_override=True
-                )
-            else:
-                raise ValueError(message)
+            docs = [orig_docstring for name in names]
+        if len(docs) != len(names):
+            message = f"""
+            Number of docs must match the number of names;
+            got {len(names)} names {tuple(names)!r} and
+            {len(docs)} docs {tuple(docs)!r}"""
+            raise ValueError(oneline(message))
+
+        can_persist = acc.can_persist
+        if can_persist is None:
+            can_persist = True
+
+        can_memoize = acc.can_memoize
+        if can_memoize is None:
+            can_memoize = True
+
+        if not can_persist and not can_memoize:
+            message = f"""
+            Attempted to set both @persist and @memoize to False.
+            At least one form of storage must be enabled for entities:
+            {provider.attrs.names!r}"""
+            raise ValueError(oneline(message))
 
         state = self._state
 
+        # Delete the original definitions.
+        # TODO Deleting the original entity definition is a little counterintuitive. It
+        # would probably make more sense to keep the original definition, although
+        # that would be a breaking change at this point. We need to think through the
+        # relationship between APIs for fixed and derived entities.
         state = state.delete_providers(provider.attrs.names)
+        state = state.delete_entity_defs(provider.attrs.names)
+
+        # Create the new definitions.
+        for name, doc in zip(names, docs):
+            entity_def = EntityDefinition(
+                name=name,
+                protocol=protocol,
+                doc=doc,
+                can_persist=can_persist,
+                can_memoize=can_memoize,
+            )
+            state = state.define_entity(entity_def)
         state = state.install_provider(provider)
 
         self._state = state
@@ -930,21 +1005,10 @@ class MergeConflict:
         self.name = name
         self.old_provider = old_state.providers_by_name.get(name)
         self.new_provider = new_state.providers_by_name.get(name)
-        self.old_is_default = name in old_state.default_provider_names
-        self.new_is_default = name in new_state.default_provider_names
-
-    @property
-    def old_protocol(self):
-        return self._protocol_for_provider(self.old_provider)
-
-    @property
-    def new_protocol(self):
-        return self._protocol_for_provider(self.new_provider)
-
-    def _protocol_for_provider(self, provider):
-        assert provider is not None
-        name_ix = provider.attrs.names.index(self.name)
-        return provider.attrs.protocols[name_ix]
+        self.old_is_default = name in old_state.default_entity_names
+        self.new_is_default = name in new_state.default_entity_names
+        self.old_entity_def = old_state.entity_defs_by_name.get(name)
+        self.new_entity_def = new_state.entity_defs_by_name.get(name)
 
     @property
     def old_is_only_declaration(self):
@@ -1016,7 +1080,7 @@ class Flow:
             The name of an entity.
         """
 
-        return self._state.get_provider(name).protocol_for_name(name)
+        return self._state.get_entity_def(name).protocol
 
     def entity_doc(self, name):
         """
@@ -1029,7 +1093,7 @@ class Flow:
         name: String
             The name of an entity.
         """
-        return self._state.get_provider(name).doc_for_name(name)
+        return self._state.get_entity_def(name).doc
 
     def entity_docstring(self, name):
         """
@@ -1682,4 +1746,4 @@ def create_default_flow_state():
         manager.start()
         return manager
 
-    return builder._state.mark_all_providers_default()
+    return builder._state.mark_all_entities_default()
diff --git a/bionic/protocols.py b/bionic/protocols.py
index 15e38f3..03dccd8 100644
--- a/bionic/protocols.py
+++ b/bionic/protocols.py
@@ -22,9 +22,8 @@ import numpy as np
 from pyarrow import parquet, Table
 import pandas as pd
 
-from .decoration import decorator_wrapping_provider
-from .exception import UnsupportedSerializedValueError
-from .provider import ProtocolUpdateProvider
+from .decoration import decorator_updating_accumulator
+from .exception import AttributeValidationError, UnsupportedSerializedValueError
 from .optdep import import_optional_dependency
 from .util import (
     read_hashable_bytes_from_file_or_dir,
@@ -186,8 +185,11 @@ class BaseProtocol:
                     )
                 )
 
-            wrapper = decorator_wrapping_provider(ProtocolUpdateProvider, self)
-            return wrapper(func)
+            return decorator_updating_accumulator(
+                lambda acc: acc.update_attr(
+                    "protocol", self, "protocol_decorator", raise_if_already_set=False
+                )
+            )(func)
         else:
             return self.__class__(**kwargs)
 
diff --git a/bionic/provider.py b/bionic/provider.py
index 80e09cb..003e1be 100644
--- a/bionic/provider.py
+++ b/bionic/provider.py
@@ -39,31 +39,17 @@ class ProviderAttributes:
     def __init__(
         self,
         names,
-        protocols=None,
         code_version=None,
         orig_flow_name=None,
-        can_persist=None,
-        can_memoize=None,
         is_default_value=None,
-        docs=None,
         changes_per_run=None,
     ):
         self.names = names
-        self.protocols = protocols
-        self.docs = docs
         self.code_version = code_version
         self.orig_flow_name = orig_flow_name
-        self._can_persist = can_persist
-        self._can_memoize = can_memoize
         self.changes_per_run = changes_per_run
         self.is_default_value = is_default_value
 
-    def should_persist(self):
-        return self._can_persist
-
-    def should_memoize(self):
-        return self._can_memoize
-
 
 class BaseProvider:
     def __init__(self, attrs):
@@ -104,30 +90,6 @@ class BaseProvider:
     def copy(self):
         raise NotImplementedError()
 
-    def protocol_for_name(self, name):
-        name_ix = self.attrs.names.index(name)
-        if name_ix < 0:
-            raise ValueError(
-                oneline(
-                    f"""
-                Attempted to look up name {name!r} from provider
-                providing only {tuple(self.attrs.names)!r}"""
-                )
-            )
-        return self.attrs.protocols[name_ix]
-
-    def doc_for_name(self, name):
-        name_ix = self.attrs.names.index(name)
-        if name_ix < 0:
-            raise ValueError(
-                oneline(
-                    f"""
-                Attempted to look up name {name!r} from provider
-                providing only {tuple(self.attrs.names)!r}"""
-                )
-            )
-        return self.attrs.docs[name_ix]
-
     def __repr__(self):
         return f"{self.__class__.__name__}{tuple(self.attrs.names)!r}"
 
@@ -141,21 +103,12 @@ class ValueProvider(BaseProvider):
             assert not provider._has_any_values
 
         names = [provider.attrs.names[0] for provider in value_providers]
-        protocols = [provider.attrs.protocols[0] for provider in value_providers]
-        docs = [provider.attrs.docs[0] for provider in value_providers]
 
-        return ValueProvider(names, protocols, docs)
+        return ValueProvider(names)
 
-    def __init__(self, names, protocols, docs):
+    def __init__(self, names):
         super(ValueProvider, self).__init__(
-            attrs=ProviderAttributes(
-                names=names,
-                protocols=protocols,
-                docs=docs,
-                can_persist=True,
-                can_memoize=True,
-                changes_per_run=False,
-            ),
+            attrs=ProviderAttributes(names=names, changes_per_run=False,),
         )
 
         self.key_space = CaseKeySpace(names)
@@ -163,9 +116,9 @@ class ValueProvider(BaseProvider):
         self._value_tuples_by_case_key = {}
         self._token_tuples_by_case_key = {}
 
-    def add_case(self, case_key, values):
+    def add_case(self, case_key, values, tokens):
         provider = self._copy()
-        provider._add_case_in_place(case_key, values)
+        provider._add_case_in_place(case_key, values, tokens)
         return provider
 
     def _copy(self):
@@ -180,12 +133,7 @@ class ValueProvider(BaseProvider):
 
     # This mutates the provider, so it should only be called on a fresh copy, as in
     # add_case().
-    def _add_case_in_place(self, case_key, values):
-        tokens = []
-        for value, protocol in zip(values, self.attrs.protocols):
-            protocol.validate(value)
-            tokens.append(protocol.tokenize(value))
-
+    def _add_case_in_place(self, case_key, values, tokens):
         if self._has_any_values:
             if case_key.space != self.key_space:
                 raise IncompatibleEntityError(
@@ -282,11 +230,7 @@ class ValueProvider(BaseProvider):
 class FunctionProvider(BaseProvider):
     def __init__(self, func):
         name = func.__name__
-        super(FunctionProvider, self).__init__(
-            attrs=ProviderAttributes(
-                names=[name], docs=(None if func.__doc__ is None else [func.__doc__]),
-            )
-        )
+        super(FunctionProvider, self).__init__(attrs=ProviderAttributes(names=[name]))
 
         self._func = func
         self.name = name
@@ -400,24 +344,6 @@ class AttrUpdateProvider(WrappingProvider):
         setattr(self.attrs, attr_name, attr_value)
 
 
-class ProtocolUpdateProvider(WrappingProvider):
-    def __init__(self, wrapped_provider, protocol=None):
-        super(ProtocolUpdateProvider, self).__init__(wrapped_provider)
-
-        protocols = [protocol for name in self.wrapped_provider.attrs.names]
-
-        self.attrs = copy(wrapped_provider.attrs)
-        self.attrs.protocols = protocols
-
-
-class MultiProtocolUpdateProvider(WrappingProvider):
-    def __init__(self, wrapped_provider, protocols=None):
-        super(MultiProtocolUpdateProvider, self).__init__(wrapped_provider)
-
-        self.attrs = copy(wrapped_provider.attrs)
-        self.attrs.protocols = protocols
-
-
 class RenamingProvider(WrappingProvider):
     def __init__(self, wrapped_provider, name):
 
@@ -475,9 +401,6 @@ class NameSplittingProvider(WrappingProvider):
 
         self.attrs = copy(wrapped_provider.attrs)
         self.attrs.names = names
-        if self.attrs.protocols is not None:
-            (protocol,) = self.attrs.protocols
-            self.attrs.protocols = [protocol for name in names]
 
     def get_tasks(self, dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode):
         inner_tasks = self.wrapped_provider.get_tasks(
diff --git a/bionic/task_state.py b/bionic/task_state.py
index f2f6056..a1fd8f0 100644
--- a/bionic/task_state.py
+++ b/bionic/task_state.py
@@ -13,17 +13,27 @@ class TaskState:
     intermediate state and the deriving logic.
     """
 
-    def __init__(self, task, dep_states, case_key, provider):
+    def __init__(self, task, dep_states, case_key, provider, entity_defs_by_name):
         self.task = task
         self.dep_states = dep_states
         self.case_key = case_key
         self.provider = provider
+        self.entity_defs_by_name = entity_defs_by_name
+
+        # In theory different entities for a single task could have different cache
+        # settings, but I'm not sure it can happen in practice (given the way grouped
+        # entities are created). At any rate, once we have tuple descriptors, each task
+        # state will only be responsible for a single entity and this won't be an issue.
+        can_persist, can_memoize = single_unique_element(
+            (entity_def.can_persist, entity_def.can_memoize)
+            for entity_def in entity_defs_by_name.values()
+        )
 
         # Cached values.
         self.task_keys = task.keys
-        self.should_memoize = provider.attrs.should_memoize()
+        self.should_memoize = can_memoize
         self.should_persist = (
-            self.provider.attrs.should_persist()
+            can_persist
             and not self.task.is_simple_lookup
             and not self.output_would_be_missing()
         )
@@ -323,9 +333,9 @@ class TaskState:
         self._queries = [
             Query(
                 task_key=task_key,
-                protocol=self.provider.protocol_for_name(
+                protocol=self.entity_defs_by_name[
                     task_key.dnode.to_entity_name()
-                ),
+                ].protocol,
                 provenance=self._provenance,
             )
             for task_key in self.task_keys


Test patch:
diff --git a/tests/test_flow/test_persistence.py b/tests/test_flow/test_persistence.py
index d181751..e7a488e 100644
--- a/tests/test_flow/test_persistence.py
+++ b/tests/test_flow/test_persistence.py
@@ -852,6 +852,19 @@ def test_disable_memory_caching(builder):
         assert flow.get("y") == 1
 
 
+@pytest.mark.parametrize("decorator", [bn.persist, bn.memoize])
+@pytest.mark.parametrize("enabled1", [True, False])
+@pytest.mark.parametrize("enabled2", [True, False])
+def test_redundant_decorators(builder, decorator, enabled1, enabled2):
+    with pytest.raises(AttributeValidationError):
+
+        @builder
+        @decorator(enabled1)
+        @decorator(enabled2)
+        def fail():
+            pass
+
+
 def test_unset_and_not_memoized(builder):
     builder.declare("x")
 
diff --git a/tests/test_flow/test_protocols.py b/tests/test_flow/test_protocols.py
index 6d78daa..79c5814 100644
--- a/tests/test_flow/test_protocols.py
+++ b/tests/test_flow/test_protocols.py
@@ -14,9 +14,13 @@ import dask.dataframe as dd
 from ..helpers import count_calls, df_from_csv_str, equal_frame_and_index_content
 
 import bionic as bn
-from bionic.util import recursively_delete_path
-from bionic.exception import UnsupportedSerializedValueError, EntitySerializationError
+from bionic.exception import (
+    AttributeValidationError,
+    EntitySerializationError,
+    UnsupportedSerializedValueError,
+)
 from bionic.protocols import CombinedProtocol, PicklableProtocol
+from bionic.util import recursively_delete_path
 
 
 PICKLABLE_VALUES = [
@@ -605,3 +609,18 @@ def test_geodataframe_protocol_fails_with_long_column_name(builder):
 
     with pytest.raises(EntitySerializationError):
         builder.build().get("longname_geopandas_df")
+
+
+@pytest.mark.parametrize("protocol1", [bn.protocol.picklable, bn.protocol.frame])
+@pytest.mark.parametrize("protocol2", [bn.protocol.picklable, bn.protocol.image])
+def test_redundant_protocols(builder, protocol1, protocol2):
+    with pytest.warns(Warning):
+
+        @builder
+        @protocol1
+        @protocol2
+        def problem():
+            return None
+
+    actual_protocol = builder.build().entity_protocol("problem")
+    assert isinstance(actual_protocol, protocol1().__class__)