Repository: square/bionic

Issue:
Errors from unset entities are hard to debug
If you try to compute an entity value but one of its ancestors doesn't have a value set, the error message is not very helpful. It just looks like this: `ValueError: Entity 'MY_ENTITY' has no defined values`. We should make it easier to figure out what the problem is and how to resolve it.

Probably the most straightforward solution would be to attach an explanation that describes which of the ancestors are not set.

Discussion:
jqmp: Looks like some of the tests don't work in parallel mode. I'll look into this.
jqmp: > Looks like some of the tests don't work in parallel mode. I'll look into this.

This is fixed.

Patch:
diff --git a/bionic/datatypes.py b/bionic/datatypes.py
index 0b5e04e..eaf1622 100644
--- a/bionic/datatypes.py
+++ b/bionic/datatypes.py
@@ -28,6 +28,18 @@ class Task:
     """
     A unit of work.  Can have dependencies, which are referred to via their
     TaskKeys.
+
+    Attributes
+    ----------
+    keys: list of TaskKeys
+        Keys corresponding to the output values computed by this task.
+    dep_keys: list of TaskKeys
+        Keys corresponding to the input values required by this task.
+    compute_func: function taking a single ``dep_values`` argument
+        Generates output values based on the passed input values.
+    is_simple_lookup: boolean
+        Whether this task consists of simply looking up the fixed value of an entity;
+        used to determine what message to log when this task is computed.
     """
 
     keys = attr.ib(converter=tuple)
@@ -83,6 +95,7 @@ class Result:
     value = attr.ib()
     file_path = attr.ib(default=None)
     value_hash = attr.ib(default=None)
+    value_is_missing = attr.ib(default=False)
 
     def __repr__(self):
         return f"Result({self.query!r}, {self.value!r})"
@@ -156,11 +169,23 @@ class CaseKeySpace(ImmutableSequence):
         return f'CaseKeySpace({", ".join(repr(name) for name in self)})'
 
 
+@attr.s(frozen=True)
+class MissingCaseKeyValue:
+    pass
+
+
 class CaseKey(ImmutableMapping):
     """
     A collection of name-value pairs that uniquely identifies a case.
     """
 
+    # This is a sentinel value used to indicate that no value is available. We can't use
+    # None because None is itself a valid value.
+    # Normally I would prefer to represent missing-ness out-of-band by making the
+    # `missing_names` field the source of truth here, but the relational methods like
+    # `project` are cleaner when we use a sentinel value.
+    MISSING = MissingCaseKeyValue()
+
     def __init__(self, name_value_token_triples):
         values_by_name = {
             name: value for name, value, token in name_value_token_triples
@@ -174,6 +199,12 @@ class CaseKey(ImmutableMapping):
         self.values = values_by_name
         self.tokens = tokens_by_name
         self.space = CaseKeySpace(list(values_by_name.keys()))
+        self.missing_names = [
+            name
+            for name, value, token in name_value_token_triples
+            if value == self.MISSING
+        ]
+        self.has_missing_values = len(self.missing_names) > 0
 
     def project(self, key_space):
         return CaseKey(
diff --git a/bionic/deriver.py b/bionic/deriver.py
index 3f7fd1e..b6c5c24 100644
--- a/bionic/deriver.py
+++ b/bionic/deriver.py
@@ -217,8 +217,14 @@ class EntityDeriver:
             for dep_dinfo in dep_dinfos
         }
 
+        # TODO Maybe of having these two separate variables that we pass around, we
+        # should just have a single method:
+        #
+        #     provider.get_dinfo(dep_dinfos_by_dnode)
         key_space = provider.get_key_space(dep_key_spaces_by_dnode)
-        tasks = provider.get_tasks(dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode)
+        tasks = provider.get_tasks(
+            dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode,
+        )
         tasks_by_key = {
             task_key: task
             for task in tasks
@@ -299,7 +305,18 @@ class EntityDeriver:
                 value; got {len(values)} ({values!r})"""
                 )
             )
-        return result_group[0].value
+        (result,) = result_group
+        if result.value_is_missing:
+            raise ValueError(
+                oneline(
+                    f"""
+                Bootstrap entity {entity_name!r} could not be computed because
+                the following entities are declared but not set:
+                {", ".join(result.query.case_key.missing_names)}
+                """
+                )
+            )
+        return result.value
 
     def _compute_result_group_for_dnode(self, dnode):
         """
@@ -383,6 +400,15 @@ class TaskKeyLogger:
 class DescriptorInfo:
     """
     Holds useful metadata about a descriptor.
+
+    Attributes
+    ----------
+    dnode: DescriptorNode
+        The descriptor this object refers to.
+    key_space: CaseKeySpace
+        Each of this descriptor's tasks' CaseKeys will have this key space.
+    tasks_by_key: dict from TaskKey to Task
+        All the tasks for this decriptor, organized by TaskKey.
     """
 
     dnode = attr.ib()
@@ -468,7 +494,7 @@ class TaskCompletionRunner:
             or self._bootstrap.process_executor is None
             # This is a non-serializable entity that needs to be returned.
             or (
-                not state.provider.attrs.should_persist()
+                not state.should_persist()
                 and entry.state.task.keys[0] in self._requested_task_keys
             )
         ):
@@ -480,7 +506,7 @@ class TaskCompletionRunner:
             self._mark_entry_completed(entry)
 
         # Process serializable entity in parallel.
-        elif state.provider.attrs.should_persist():
+        elif state.should_persist():
             # TODO: Logging support for multiple processes not done yet.
             new_state_for_subprocess = state.strip_state_for_subprocess()
             future = self._bootstrap.process_executor.submit(
diff --git a/bionic/exception.py b/bionic/exception.py
index 2869d43..0a9a22e 100644
--- a/bionic/exception.py
+++ b/bionic/exception.py
@@ -15,6 +15,10 @@ class AlreadyDefinedEntityError(ValueError):
         return cls(f"Entity {name!r} is already defined")
 
 
+class UnsetEntityError(ValueError):
+    pass
+
+
 class IncompatibleEntityError(ValueError):
     pass
 
diff --git a/bionic/flow.py b/bionic/flow.py
index aec1ae9..fa2008d 100644
--- a/bionic/flow.py
+++ b/bionic/flow.py
@@ -19,11 +19,12 @@ 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
+from .datatypes import CaseKey, VersioningPolicy, ResultGroup
 from .exception import (
     UndefinedEntityError,
     AlreadyDefinedEntityError,
     IncompatibleEntityError,
+    UnsetEntityError,
 )
 from .optdep import import_optional_dependency
 from .provider import (
@@ -1102,7 +1103,14 @@ class Flow:
         """
 
         dnode = entity_dnode_from_descriptor(name)
-        result_group = self._deriver.derive(dnode)
+        orig_result_group = self._deriver.derive(dnode)
+        # Remove all results with missing values.
+        result_group = ResultGroup(
+            results=[
+                result for result in orig_result_group if not result.value_is_missing
+            ],
+            key_space=orig_result_group.key_space,
+        )
 
         if mode is object or mode == "object":
             values = [result.value for result in result_group]
@@ -1140,7 +1148,17 @@ class Flow:
 
         if collection is None or collection is object:
             if len(values) == 0:
-                raise ValueError(f"Entity {name!r} has no defined values")
+                # There should always be at least one result, but some of them may have
+                # missing values.
+                assert len(orig_result_group) > 0
+                missing_result = orig_result_group[0]
+                missing_names = missing_result.query.case_key.missing_names
+                message = f"""
+                Entity {name} could not be computed because the following entities are
+                declared but have no values set:
+                {", ".join(repr(name) for name in missing_names)}
+                """
+                raise UnsetEntityError(oneline(message))
             if len(values) > 1:
                 raise ValueError(f"Entity {name!r} has multiple values")
             return values[0]
@@ -1160,7 +1178,6 @@ class Flow:
         else:
             raise ValueError(f"Unrecognized collection type {collection!r}")
 
-    # TODO Maybe this wants to be two different functions?
     def export(self, name, file_path=None, dir_path=None):
         """
         Provides access to the persisted file corresponding to an entity.  Note:
diff --git a/bionic/provider.py b/bionic/provider.py
index 441c296..061531d 100644
--- a/bionic/provider.py
+++ b/bionic/provider.py
@@ -14,6 +14,7 @@ import functools
 from io import BytesIO
 
 import pandas as pd
+import attr
 
 from .datatypes import (
     Task,
@@ -157,7 +158,7 @@ class ValueProvider(BaseProvider):
             ),
         )
 
-        self.key_space = CaseKeySpace()
+        self.key_space = CaseKeySpace(names)
         self._has_any_values = False
         self._value_tuples_by_case_key = {}
         self._token_tuples_by_case_key = {}
@@ -215,7 +216,11 @@ class ValueProvider(BaseProvider):
         return self._has_any_values
 
     def get_code_fingerprint(self, case_key):
-        value_tokens = " ".join(self._token_tuples_by_case_key[case_key])
+        if not self.has_any_cases():
+            value_tokens = ""
+        else:
+            value_tokens = " ".join(self._token_tuples_by_case_key[case_key])
+
         return CodeFingerprint(
             version=CodeVersion(major=value_tokens, minor=None,),
             bytecode_hash=None,
@@ -233,18 +238,42 @@ class ValueProvider(BaseProvider):
         assert not dep_key_spaces_by_dnode
         assert not dep_task_key_lists_by_dnode
 
-        return [
-            Task(
-                keys=[
-                    TaskKey(entity_dnode_from_descriptor(name), case_key)
-                    for name in self.attrs.names
-                ],
-                dep_keys=[],
-                compute_func=functools.partial(self._compute, case_key=case_key,),
-                is_simple_lookup=True,
-            )
-            for case_key in self._value_tuples_by_case_key.keys()
-        ]
+        if self.has_any_cases():
+            return [
+                Task(
+                    keys=[
+                        TaskKey(
+                            dnode=entity_dnode_from_descriptor(name), case_key=case_key,
+                        )
+                        for name in self.attrs.names
+                    ],
+                    dep_keys=[],
+                    compute_func=functools.partial(self._compute, case_key=case_key,),
+                    is_simple_lookup=True,
+                )
+                for case_key in self._value_tuples_by_case_key.keys()
+            ]
+
+        # If we have no cases, we instead return a single "missing value" task.
+        else:
+            return [
+                Task(
+                    keys=[
+                        TaskKey(
+                            dnode=entity_dnode_from_descriptor(name),
+                            case_key=CaseKey(
+                                [
+                                    (name, CaseKey.MISSING, "<MISSING>")
+                                    for name in self.attrs.names
+                                ]
+                            ),
+                        )
+                        for name in self.attrs.names
+                    ],
+                    dep_keys=[],
+                    compute_func=None,
+                )
+            ]
 
     def _compute(self, dep_values, case_key):
         return self._value_tuples_by_case_key[case_key]
@@ -292,10 +321,11 @@ class FunctionProvider(BaseProvider):
 
         return [
             Task(
-                keys=[TaskKey(self.dnode, case_key)],
+                keys=[TaskKey(dnode=self.dnode, case_key=case_key)],
                 dep_keys=[
                     TaskKey(
-                        dep_dnode, case_key.project(dep_key_spaces_by_dnode[dep_dnode]),
+                        dnode=dep_dnode,
+                        case_key=case_key.project(dep_key_spaces_by_dnode[dep_dnode]),
                     )
                     for dep_dnode in self._dep_dnodes
                 ],
@@ -340,7 +370,7 @@ class WrappingProvider(BaseProvider):
 
     def get_tasks(self, dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode):
         return self.wrapped_provider.get_tasks(
-            dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode
+            dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode,
         )
 
     def get_source_func(self):
@@ -423,7 +453,7 @@ class RenamingProvider(WrappingProvider):
             )
 
         inner_tasks = self.wrapped_provider.get_tasks(
-            dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode
+            dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode,
         )
         return [wrap_task(task) for task in inner_tasks]
 
@@ -451,7 +481,7 @@ class NameSplittingProvider(WrappingProvider):
 
     def get_tasks(self, dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode):
         inner_tasks = self.wrapped_provider.get_tasks(
-            dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode
+            dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode,
         )
 
         def wrap_task(task):
@@ -578,22 +608,53 @@ class GatherProvider(WrappingProvider):
         ]
         gather_case_keys = merge_case_key_lists(gather_case_key_lists)
 
-        gather_case_key_lists_by_delta_case_key = defaultdict(list)
+        # For each of these case keys, collect a "row" of the associated dependency
+        # task keys.
+        gather_rows = []
         for gather_case_key in gather_case_keys:
-            delta_case_key = gather_case_key.project(key_spaces.delta)
-            gather_case_key_lists_by_delta_case_key[delta_case_key].append(
-                gather_case_key
+            dep_task_keys = []
+            for dep_dnode in self._gather_dnodes:
+                dep_key_space = dep_key_spaces_by_dnode[dep_dnode]
+                dep_case_key = gather_case_key.project(dep_key_space)
+                dep_task_key = TaskKey(dep_dnode, dep_case_key)
+                dep_task_keys.append(dep_task_key)
+
+            gather_row = GatherRow(
+                dep_task_keys=dep_task_keys, full_case_key=gather_case_key,
             )
+            gather_rows.append(gather_row)
 
-        delta_case_keys = gather_case_key_lists_by_delta_case_key.keys()
+        # Now partition these rows into tables. Each table corresponds to one gathered
+        # dataframe.
+        gather_row_lists_by_delta_case_key = groups_dict(
+            gather_rows, lambda row: row.full_case_key.project(key_spaces.delta)
+        )
+        gather_tables_by_out_task_key = {}
+        for (
+            delta_case_key,
+            all_gather_rows,
+        ) in gather_row_lists_by_delta_case_key.items():
+            gather_rows_without_missing_case_key_values = [
+                gather_row
+                for gather_row in all_gather_rows
+                if not any(
+                    dep_task_key.case_key.project(key_spaces.primary).has_missing_values
+                    for dep_task_key in gather_row.dep_task_keys
+                )
+            ]
+            table_task_key = TaskKey(
+                dnode=self._inner_gathered_dep_dnode, case_key=delta_case_key,
+            )
+            gather_table = GatherTable(
+                rows=gather_rows_without_missing_case_key_values,
+                out_task_key=table_task_key,
+            )
+            gather_tables_by_out_task_key[table_task_key] = gather_table
 
-        # Create new task keys for the gathered values that the inner provider
+        # Create new task keys for the gathered frame values that the inner provider
         # will consume.
         inner_gathered_key_space = key_spaces.delta
-        inner_gathered_dep_task_keys = [
-            TaskKey(dnode=self._inner_gathered_dep_dnode, case_key=case_key,)
-            for case_key in delta_case_keys
-        ]
+        inner_gathered_dep_task_keys = list(gather_tables_by_out_task_key.keys())
 
         # Now we can construct the dicts of key spaces and task keys that the
         # wrapper provider will see.
@@ -605,7 +666,6 @@ class GatherProvider(WrappingProvider):
             )
             for dnode in self._inner_dep_dnodes
         }
-
         inner_dtkls_by_dnode = {
             dnode: (
                 inner_gathered_dep_task_keys
@@ -629,26 +689,10 @@ class GatherProvider(WrappingProvider):
             # Remove the key for the aggregated value, and remember its case
             # key.
             passthrough_dep_keys = list(inner_dep_keys)
-            inner_gather_case_key = passthrough_dep_keys.pop(
-                gather_task_key_ix
-            ).case_key
-            delta_case_key = inner_gather_case_key
-
-            # Find the task keys that need to be gathered together.
-            unique_gather_task_keys = set()
-            for dep_dnode in self._gather_dnodes:
-                dep_key_space = dep_key_spaces_by_dnode[dep_dnode]
+            inner_gather_task_key = passthrough_dep_keys.pop(gather_task_key_ix)
 
-                relevant_gather_case_keys = gather_case_key_lists_by_delta_case_key[
-                    delta_case_key
-                ]
-                for gather_case_key in relevant_gather_case_keys:
-                    unique_gather_task_keys.add(
-                        TaskKey(
-                            dnode=dep_dnode,
-                            case_key=gather_case_key.project(dep_key_space),
-                        )
-                    )
+            # Identify the table that this task corresponds to.
+            gather_table = gather_tables_by_out_task_key[inner_gather_task_key]
 
             # NOTE prepended_keys has a non-deterministic order, because a
             # set's ordering depends on the hashes of its contents, and TaskKey
@@ -659,7 +703,12 @@ class GatherProvider(WrappingProvider):
             # to eliminate every source of non-deterministic ordering, I think
             # it's better to sort our values at the point where we actually
             # need determinism.  (Viz., computing the provenance hash.)
-            prepended_keys = list(unique_gather_task_keys)
+            unique_gather_dep_keys = set(
+                dep_task_key
+                for gather_row in gather_table.rows
+                for dep_task_key in gather_row.dep_task_keys
+            )
+            prepended_keys = list(unique_gather_dep_keys)
 
             # Combine the gathering task keys with the keys expected by the
             # wrapped task (except the one key we removed, since we'll be
@@ -674,17 +723,11 @@ class GatherProvider(WrappingProvider):
                 values_by_task_key = dict(list(zip(prepended_keys, prepended_values)))
 
                 # Gather the prepended values into a single frame.
-                row_case_keys = gather_case_key_lists_by_delta_case_key[delta_case_key]
-
                 gathered_df = pd.DataFrame()
-                for dnode in self._gather_dnodes:
-                    key_space = dep_key_spaces_by_dnode[dnode]
-                    descriptor = dnode.to_descriptor()
-                    gathered_df[descriptor] = [
-                        values_by_task_key.get(
-                            TaskKey(dnode, case_key.project(key_space)), None
-                        )
-                        for case_key in row_case_keys
+                for dep_ix, dnode in enumerate(self._gather_dnodes):
+                    gathered_df[dnode.to_descriptor()] = [
+                        values_by_task_key.get(gather_row.dep_task_keys[dep_ix], None)
+                        for gather_row in gather_table.rows
                     ]
 
                 # Construct the final values to pass to the wrapped task.
@@ -704,7 +747,7 @@ class GatherProvider(WrappingProvider):
             )
 
         orig_tasks = self.wrapped_provider.get_tasks(
-            inner_key_spaces_by_dnode, inner_dtkls_by_dnode
+            inner_key_spaces_by_dnode, inner_dtkls_by_dnode,
         )
         return [wrap_task(task) for task in orig_tasks]
 
@@ -792,21 +835,19 @@ class PyplotProvider(WrappingProvider):
     def get_key_space(self, dep_key_spaces_by_dnode):
         outer_dkss = dep_key_spaces_by_dnode
         inner_dkss = outer_dkss.copy()
-        inner_dkss[self._pyplot_dnode] = CaseKey([])
+        inner_dkss[self._pyplot_dnode] = CaseKeySpace()
 
         return self.wrapped_provider.get_key_space(inner_dkss)
 
     def get_tasks(self, dep_key_spaces_by_dnode, dep_task_key_lists_by_dnode):
         outer_dkss = dep_key_spaces_by_dnode
-
         inner_dkss = outer_dkss.copy()
-        inner_dkss[self._pyplot_dnode] = CaseKey([])
+        inner_dkss[self._pyplot_dnode] = CaseKeySpace()
 
         outer_dtkls = dep_task_key_lists_by_dnode
-
         inner_dtkls = outer_dtkls.copy()
         inner_dtkls[self._pyplot_dnode] = [
-            TaskKey(dnode=self._pyplot_dnode, case_key=CaseKey([]),)
+            TaskKey(dnode=self._pyplot_dnode, case_key=CaseKey([]))
         ]
 
         inner_tasks = self.wrapped_provider.get_tasks(inner_dkss, inner_dtkls)
@@ -955,6 +996,28 @@ def multi_index_from_case_keys(case_keys, ordered_key_names):
     )
 
 
+@attr.s
+class GatherRow:
+    """
+    A collection of task keys corresponding to one row of a gathered dataframe.
+    """
+
+    dep_task_keys = attr.ib()
+    full_case_key = attr.ib()
+
+
+@attr.s
+class GatherTable:
+    """
+    Holds the task keys for one gathered dataframe. Contains the task keys required
+    to compute each cell (organized by row) and one task key corresponding to the
+    assembled frame itself.
+    """
+
+    rows = attr.ib()
+    out_task_key = attr.ib()
+
+
 # -- Helpers for working with providers.
 
 PROVIDER_METHODS = [
diff --git a/bionic/task_state.py b/bionic/task_state.py
index b247bc9..381f868 100644
--- a/bionic/task_state.py
+++ b/bionic/task_state.py
@@ -3,7 +3,7 @@ import copy
 from .cache import Provenance
 from .datatypes import ProvenanceDigest, Query, Result
 from .exception import CodeVersioningError
-from .util import oneline
+from .util import oneline, single_unique_element
 
 
 class TaskState:
@@ -28,7 +28,7 @@ class TaskState:
         # This can be set by complete() or _compute().
         #
         # This will be present if and only if both is_complete and
-        # provider.attrs.should_persist() are True.
+        # should_persist() are True.
         self._result_value_hashes_by_name = None
 
         # This can be set by get_results_assuming_complete() or _compute().
@@ -57,7 +57,7 @@ class TaskState:
         return [
             dep_state
             for dep_state in self.dep_states
-            if (dep_state.provider.attrs.should_persist() and not dep_state.is_complete)
+            if (dep_state.should_persist() and not dep_state.is_complete)
             or not dep_state.is_completable
         ]
 
@@ -77,6 +77,19 @@ class TaskState:
         """
         return self._is_initialized and len(self.blocking_dep_states()) == 0
 
+    # TODO Now that we've introduced this method, let's remove
+    # provider.attrs.should_persist(), which is just a wrapper for
+    # provider.attrs._can_persist anyway.
+    def should_persist(self):
+        return (
+            self.provider.attrs.should_persist() and not self.output_would_be_missing()
+        )
+
+    def output_would_be_missing(self):
+        return single_unique_element(
+            task_key.case_key.has_missing_values for task_key in self.task.keys
+        )
+
     def __repr__(self):
         return f"TaskState({self.task!r})"
 
@@ -91,7 +104,7 @@ class TaskState:
         assert not self.is_complete
 
         # See if we can load it from the cache.
-        if self.provider.attrs.should_persist() and all(
+        if self.should_persist() and all(
             axr.can_load() for axr in self._cache_accessors
         ):
             # We only load the hashed result while completing task state
@@ -150,15 +163,15 @@ class TaskState:
         for dep_state, dep_key in zip(self.dep_states, task.dep_keys):
             assert dep_state.is_complete or dep_state.is_completable
             if dep_state.is_complete:
-                dep_result_by_name = dep_state.get_results_assuming_complete(
+                dep_results_by_name = dep_state.get_results_assuming_complete(
                     task_key_logger
                 )
             else:
                 # If dep_state is not complete yet, that's probably because the results
                 # aren't communicated between processes. Compute the results to populate
                 # the in-memory cache.
-                dep_result_by_name = dep_state._compute(task_key_logger)
-            dep_results.append(dep_result_by_name[dep_key.dnode.to_entity_name()])
+                dep_results_by_name = dep_state._compute(task_key_logger)
+            dep_results.append(dep_results_by_name[dep_key.dnode.to_entity_name()])
 
         provider = self.provider
 
@@ -168,6 +181,26 @@ class TaskState:
 
         dep_values = [dep_result.value for dep_result in dep_results]
 
+        # If we have any missing outputs, exit early with a missing result.
+        if self.output_would_be_missing():
+            results_by_name = {}
+            result_value_hashes_by_name = {}
+            for query in self._queries:
+                result = Result(query=query, value=None, value_is_missing=True)
+                entity_name = query.dnode.to_entity_name()
+                results_by_name[entity_name] = result
+                result_value_hashes_by_name[entity_name] = ""
+            self._results_by_name = results_by_name
+            self._result_value_hashes_by_name = result_value_hashes_by_name
+            return self._results_by_name
+
+        else:
+            # If we have no missing outputs, we should not be consuming any missing
+            # inputs either.
+            assert not any(
+                dep_key.case_key.has_missing_values for dep_key in task.dep_keys
+            )
+
         values = task.compute(dep_values)
         assert len(values) == len(provider.attrs.names)
 
@@ -182,9 +215,9 @@ class TaskState:
         for ix, (query, value) in enumerate(zip(self._queries, values)):
             query.protocol.validate(value)
 
-            result = Result(query=query, value=value,)
+            result = Result(query=query, value=value)
 
-            if provider.attrs.should_persist():
+            if self.should_persist():
                 accessor = self._cache_accessors[ix]
                 accessor.save_result(result)
 
@@ -197,11 +230,11 @@ class TaskState:
         # Otherwise, load it lazily later so that if the serialized/deserialized
         # value is not exactly the same as the original, we still
         # always return the same value.
-        if provider.attrs.should_memoize() and not provider.attrs.should_persist():
+        if provider.attrs.should_memoize() and not self.should_persist():
             self._results_by_name = results_by_name
 
         # But we cache the hashed values eagerly since they are cheap to load.
-        if provider.attrs.should_persist():
+        if self.should_persist():
             self._result_value_hashes_by_name = result_value_hashes_by_name
 
         return self._results_by_name
@@ -215,7 +248,7 @@ class TaskState:
         back from the subprocess.
         """
 
-        assert self.provider.attrs.should_persist()
+        assert self.should_persist()
 
         # First, let's flush the stored entries in cache accessors. This is to prevent cases
         # like current process contains an outdated stored entries.
@@ -267,7 +300,7 @@ class TaskState:
         dep_provenance_digests_by_task_key = {}
         for dep_key, dep_state in zip(self.task.dep_keys, self.dep_states):
             # Use value hash of persistable values.
-            if dep_state.provider.attrs.should_persist():
+            if dep_state.should_persist():
                 value_hash = dep_state._result_value_hashes_by_name[
                     dep_key.dnode.to_entity_name()
                 ]
@@ -302,7 +335,7 @@ class TaskState:
         ]
 
         # Lastly, set up cache accessors.
-        if self.provider.attrs.should_persist():
+        if self.should_persist():
             if bootstrap is None:
                 name = self.task.keys[0].entity_name
                 raise AssertionError(
@@ -408,7 +441,7 @@ class TaskState:
 
         # Any non-persisted value will have to be recomputed again in the other
         # subprocess and isn't complete anymore.
-        if not task_state.provider.attrs.should_persist():
+        if not task_state.should_persist():
             task_state.is_complete = False
 
         if task_state.is_complete:
diff --git a/bionic/util.py b/bionic/util.py
index 3f5a11a..de0739f 100644
--- a/bionic/util.py
+++ b/bionic/util.py
@@ -78,7 +78,14 @@ def single_element(iterable):
         )
     return items[0]
 
-    return next(iter(iterable))
+
+def single_unique_element(iterable):
+    """
+    Takes an iterable with exactly one unique element and returns that element.
+
+    Equivalent to `single_element(set(iterable))`.
+    """
+    return single_element(set(iterable))
 
 
 def hash_to_hex(bytestring, n_bytes=None):


Test patch:
diff --git a/tests/test_flow/test_api.py b/tests/test_flow/test_api.py
index 4eb2ac1..02c23f0 100644
--- a/tests/test_flow/test_api.py
+++ b/tests/test_flow/test_api.py
@@ -16,6 +16,7 @@ from bionic.exception import (
     EntitySerializationError,
     IncompatibleEntityError,
     UndefinedEntityError,
+    UnsetEntityError,
 )
 
 from ..helpers import assert_re_matches, count_calls
@@ -335,15 +336,17 @@ def test_merge(builder):
 def test_get_single(preset_flow):
     flow = preset_flow
 
-    with raises(ValueError):
+    with raises(UnsetEntityError):
         flow.get("x")
 
     assert flow.get("y") == 1
 
     with raises(ValueError):
         assert flow.get("z")
-    with raises(ValueError):
+
+    with raises(UnsetEntityError) as excinfo:
         assert flow.get("f")
+    assert "'x'" in str(excinfo.value)
 
     assert flow.get("p") == 4
     assert flow.get("q") == 5
diff --git a/tests/test_flow/test_gather.py b/tests/test_flow/test_gather.py
index 2edbdfa..6ee4bae 100644
--- a/tests/test_flow/test_gather.py
+++ b/tests/test_flow/test_gather.py
@@ -44,6 +44,36 @@ def test_x(preset_builder):
     }
 
 
+def test_x__with_empty_x(preset_builder):
+    builder = preset_builder
+
+    builder.set("x", values=[])
+
+    @builder
+    @bn.gather("x")
+    def summary(gather_df):
+        return summarize(gather_df)
+
+    assert builder.build().get("summary", set) == {
+        "",
+    }
+
+
+def test_x__with_unset_x(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("x")
+
+    @builder
+    @bn.gather("x")
+    def summary(gather_df):
+        return summarize(gather_df)
+
+    assert builder.build().get("summary", list) == [
+        "",
+    ]
+
+
 def test_x__y(preset_builder):
     builder = preset_builder
 
@@ -58,6 +88,36 @@ def test_x__y(preset_builder):
     }
 
 
+def test_x__y__with_unset_x(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("x")
+
+    @builder
+    @bn.gather("x", "y")
+    def summary(gather_df):
+        assert set(gather_df.columns) == {"x", "y"}
+        return summarize(gather_df)
+
+    assert builder.build().get("summary", list) == [
+        "",
+        "",
+    ]
+
+
+def test_x__y__with_unset_y(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("y")
+
+    @builder
+    @bn.gather("x", "y")
+    def summary(gather_df):
+        return summarize(gather_df)
+
+    assert builder.build().get("summary", set) == set()
+
+
 def test_x__xy(preset_builder):
     builder = preset_builder
 
@@ -72,6 +132,35 @@ def test_x__xy(preset_builder):
     }
 
 
+def test_x__xy__with_unset_x(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("x")
+
+    @builder
+    @bn.gather("x", "xy")
+    def summary(gather_df):
+        return summarize(gather_df)
+
+    assert builder.build().get("summary", list) == [
+        "",
+        "",
+    ]
+
+
+def test_x__xy__with_unset_y(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("y")
+
+    @builder
+    @bn.gather("x", "xy")
+    def summary(gather_df):
+        return summarize(gather_df)
+
+    assert builder.build().get("summary", list) == []
+
+
 def test_xy__yz(preset_builder):
     builder = preset_builder
 
@@ -86,6 +175,51 @@ def test_xy__yz(preset_builder):
     }
 
 
+def test_xy__yz__with_unset_x(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("x")
+
+    @builder
+    @bn.gather("xy", "yz")
+    def summary(gather_df):
+        return summarize(gather_df)
+
+    assert builder.build().get("summary", list) == [
+        "",
+        "",
+    ]
+
+
+def test_xy__yz__with_unset_y(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("y")
+
+    @builder
+    @bn.gather("xy", "yz")
+    def summary(gather_df):
+        return summarize(gather_df)
+
+    assert builder.build().get("summary", list) == [
+        "",
+        "",
+    ]
+
+
+def test_xy__yz__with_unset_z(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("z")
+
+    @builder
+    @bn.gather("xy", "yz")
+    def summary(gather_df):
+        return summarize(gather_df)
+
+    assert builder.build().get("summary", list) == []
+
+
 def test_x_y__z(preset_builder):
     builder = preset_builder
 
@@ -146,6 +280,19 @@ def test_wx__xy__z(preset_builder):
     }
 
 
+def test_wx__xy__z_with_unset_z(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("z")
+
+    @builder
+    @bn.gather("wx", "xy")
+    def summary(gather_df, z):
+        return summarize(gather_df) + " -- " + z
+
+    assert builder.build().get("summary", set) == set()
+
+
 def test_wx__xy__yz(preset_builder):
     builder = preset_builder
 
@@ -182,6 +329,63 @@ def test_xy__yz__wx(preset_builder):
     }
 
 
+def test_xy__yz__wx__unset_x(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("x")
+
+    @builder
+    @bn.gather("xy", "yz")
+    def summary(gather_df, wx):
+        return summarize(gather_df) + " -- " + wx
+
+    assert builder.build().get("summary", list) == []
+
+
+def test_xy__yz__wx__unset_y(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("y")
+
+    @builder
+    @bn.gather("xy", "yz")
+    def summary(gather_df, wx):
+        return summarize(gather_df) + " -- " + wx
+
+    assert builder.build().get("summary", set) == {
+        " -- WX",
+        " -- Wx",
+        " -- wX",
+        " -- wx",
+    }
+
+
+def test_xy__yz__wx__unset_z(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("z")
+
+    @builder
+    @bn.gather("xy", "yz")
+    def summary(gather_df, wx):
+        return summarize(gather_df) + " -- " + wx
+
+    assert builder.build().get("summary", list) == []
+
+
+def test_xy__yz__wx__unset_w(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("w")
+
+    @builder
+    @bn.gather("xy", "yz")
+    def summary(gather_df, wx):
+        return summarize(gather_df) + " -- " + wx
+
+    assert builder.build().get("summary", list) == []
+
+
 def test_rename_frame(preset_builder):
     builder = preset_builder
 
@@ -213,6 +417,72 @@ def test_multiple_gathers(preset_builder):
     }
 
 
+def test_multiple_gathers__unset_x(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("x")
+
+    @builder
+    @bn.gather("w", "wx", "df1")
+    @bn.gather("y", "yz", "df2")
+    def summary(df1, df2):
+        return summarize(df1) + " -- " + summarize(df2)
+
+    assert builder.build().get("summary", list) == []
+
+
+def test_multiple_gathers__unset_y(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("y")
+
+    @builder
+    @bn.gather("w", "wx", "df1")
+    @bn.gather("y", "yz", "df2")
+    def summary(df1, df2):
+        return summarize(df1) + " -- " + summarize(df2)
+
+    assert builder.build().get("summary", set) == {
+        "W.WX w.wX -- ",
+        "W.WX w.wX -- ",
+        "W.Wx w.wx -- ",
+        "W.Wx w.wx -- ",
+    }
+
+
+def test_multiple_gathers__unset_z(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("z")
+
+    @builder
+    @bn.gather("w", "wx", "df1")
+    @bn.gather("y", "yz", "df2")
+    def summary(df1, df2):
+        return summarize(df1) + " -- " + summarize(df2)
+
+    assert builder.build().get("summary", list) == []
+
+
+def test_multiple_gathers__unset_w(preset_builder):
+    builder = preset_builder
+
+    builder.clear_cases("w")
+
+    @builder
+    @bn.gather("w", "wx", "df1")
+    @bn.gather("y", "yz", "df2")
+    def summary(df1, df2):
+        return summarize(df1) + " -- " + summarize(df2)
+
+    assert builder.build().get("summary", set) == {
+        " -- Y.YZ y.yZ",
+        " -- Y.Yz y.yz",
+        " -- Y.YZ y.yZ",
+        " -- Y.Yz y.yz",
+    }
+
+
 def test_multiple_gathers_complex(preset_builder):
     builder = preset_builder
 
diff --git a/tests/test_flow/test_join.py b/tests/test_flow/test_join.py
index 395b1e0..49e3429 100644
--- a/tests/test_flow/test_join.py
+++ b/tests/test_flow/test_join.py
@@ -1,5 +1,7 @@
 import pytest
 
+from bionic.exception import UnsetEntityError
+
 
 @pytest.fixture(scope="function")
 def preset_builder(builder):
@@ -75,3 +77,8 @@ def test_empty(preset_builder):
     assert flow.get("xy", set) == set()
     assert flow.get("yz", set) == {12, 15}
     assert flow.get("xy_plus_yz", set) == set()
+
+    with pytest.raises(UnsetEntityError):
+        flow.get("xy")
+    with pytest.raises(UnsetEntityError):
+        flow.get("xy_plus_yz")
diff --git a/tests/test_flow/test_persistence.py b/tests/test_flow/test_persistence.py
index 02585ad..d181751 100644
--- a/tests/test_flow/test_persistence.py
+++ b/tests/test_flow/test_persistence.py
@@ -852,6 +852,28 @@ def test_disable_memory_caching(builder):
         assert flow.get("y") == 1
 
 
+def test_unset_and_not_memoized(builder):
+    builder.declare("x")
+
+    @builder
+    @bn.memoize(False)
+    def x_plus_one(x):
+        return x + 1
+
+    assert builder.build().get("x_plus_one", list) == []
+
+
+def test_unset_and_not_persisted(builder):
+    builder.declare("x")
+
+    @builder
+    @bn.persist(False)
+    def x_plus_one(x):
+        return x + 1
+
+    assert builder.build().get("x_plus_one", list) == []
+
+
 @pytest.mark.no_parallel
 def test_changes_per_run_and_not_persist(builder, make_counter):
     builder.assign("x", 5)