Simplify relation tree structure.

This merges DimensionJoin, Selection, and OrderedSlice into a single
leaf-or-join Select relation, while restricting the relation types that
can appear upstream or downstream of it.  This means almost all trees
will be structured as (with [] meaning optional):

                                          ╭─ [DatasetSearch]
[OrderedSlice] <─ [FindFirst] <─ Select <─┼─ [DataCoordinateUpload]
                                          ╰─ [Materialization] <─ ...

where "..." is another tree of the same structure.  While this *can*
do arbitrary nesting, we never descend past a Materialization when
figuring out where put a new operation when pushing a new filter or
join onto the tree, so all the logic for that can just work with the
first Select it sees rather than worry about what's upstream.  It's
hard to overstate how much of a simplification this is going to be
compared to the more general daf_relation trees.

The downside of this restricted-tree approach is that it obfuscates
what happens in SQL vs. Python postprocessing, and while that might
seem like a good thing (since it *mostly*) is, the performance
difference involved in moving certain operations down to Python can be
so huge that we can't pretend it's a pure implementation, and this
simplified tree structure may make it a bit more difficult to warn or
fail if the user tries to do something that *seems* reasonable but
actually isn't when those implementation/performance implications are
in play.

python/lsst/daf/butler/queries/_query.py

@@ -36,7 +36,7 @@
 from ..dimensions import DataCoordinate, DataId, DimensionGroup
 from .data_coordinate_results import DataCoordinateResultSpec, RelationDataCoordinateQueryResults
 from .driver import QueryDriver
-from .relation_tree import Relation
+from .relation_tree import RootRelation
 
 if TYPE_CHECKING:
     from .._query_results import DataCoordinateQueryResults, DatasetQueryResults, DimensionRecordQueryResults
@@ -65,7 +65,7 @@ class RelationQuery(Query):
     if this is the only implementation.
     """
 
-    def __init__(self, driver: QueryDriver, tree: Relation, include_dimension_records: bool):
+    def __init__(self, driver: QueryDriver, tree: RootRelation, include_dimension_records: bool):
         self._driver = driver
         self._tree = tree
         self._include_dimension_records = include_dimension_records

python/lsst/daf/butler/queries/data_coordinate_results.py

@@ -44,7 +44,7 @@
 from .._query_results import DataCoordinateQueryResults, DatasetQueryResults
 from ..dimensions import DataCoordinate, DimensionGroup
 from .driver import QueryDriver
-from .relation_tree import Relation
+from .relation_tree import RootRelation
 
 if TYPE_CHECKING:
     from .driver import PageKey
@@ -82,7 +82,7 @@ class RelationDataCoordinateQueryResults(DataCoordinateQueryResults):
     we won't need an ABC if this is the only implementation.
     """
 
-    def __init__(self, tree: Relation, driver: QueryDriver, spec: DataCoordinateResultSpec):
+    def __init__(self, tree: RootRelation, driver: QueryDriver, spec: DataCoordinateResultSpec):
         self._tree = tree
         self._driver = driver
         self._spec = spec

python/lsst/daf/butler/queries/driver.py

@@ -42,7 +42,7 @@
 from .dataset_results import DatasetRefResultPage, DatasetRefResultSpec
 from .dimension_record_results import DimensionRecordResultPage, DimensionRecordResultSpec
 from .general_results import GeneralResultPage, GeneralResultSpec
-from .relation_tree import Relation
+from .relation_tree import RootRelation
 
 PageKey: TypeAlias = uuid.UUID
 
@@ -89,23 +89,25 @@ def universe(self) -> DimensionUniverse:
         raise NotImplementedError()
 
     @overload
-    def execute(self, tree: Relation, result_spec: DataCoordinateResultSpec) -> DataCoordinateResultPage:
+    def execute(self, tree: RootRelation, result_spec: DataCoordinateResultSpec) -> DataCoordinateResultPage:
         ...
 
     @overload
-    def execute(self, tree: Relation, result_spec: DimensionRecordResultSpec) -> DimensionRecordResultPage:
+    def execute(
+        self, tree: RootRelation, result_spec: DimensionRecordResultSpec
+    ) -> DimensionRecordResultPage:
         ...
 
     @overload
-    def execute(self, tree: Relation, result_spec: DatasetRefResultSpec) -> DatasetRefResultPage:
+    def execute(self, tree: RootRelation, result_spec: DatasetRefResultSpec) -> DatasetRefResultPage:
         ...
 
     @overload
-    def execute(self, tree: Relation, result_spec: GeneralResultSpec) -> GeneralResultPage:
+    def execute(self, tree: RootRelation, result_spec: GeneralResultSpec) -> GeneralResultPage:
         ...
 
     @abstractmethod
-    def execute(self, tree: Relation, result_spec: ResultSpec) -> ResultPage:
+    def execute(self, tree: RootRelation, result_spec: ResultSpec) -> ResultPage:
         """Execute a query and return the first result page.
 
         Parameters
@@ -184,7 +186,7 @@ def fetch_next_page(self, result_spec: ResultSpec, key: PageKey) -> ResultPage:
         raise NotImplementedError()
 
     @abstractmethod
-    def count(self, tree: Relation, *, exact: bool, discard: bool) -> int:
+    def count(self, tree: RootRelation, *, exact: bool, discard: bool) -> int:
         """Return the number of rows a query would return.
 
         Parameters
@@ -206,7 +208,7 @@ def count(self, tree: Relation, *, exact: bool, discard: bool) -> int:
         raise NotImplementedError()
 
     @abstractmethod
-    def any(self, tree: Relation, *, execute: bool, exact: bool) -> bool:
+    def any(self, tree: RootRelation, *, execute: bool, exact: bool) -> bool:
         """Test whether the query would return any rows.
 
         Parameters
@@ -232,7 +234,7 @@ def any(self, tree: Relation, *, execute: bool, exact: bool) -> bool:
         raise NotImplementedError()
 
     @abstractmethod
-    def explain_no_results(self, tree: Relation, execute: bool) -> Iterable[str]:
+    def explain_no_results(self, tree: RootRelation, execute: bool) -> Iterable[str]:
         """Return human-readable messages that may help explain why the query
         yields no results.
 

python/lsst/daf/butler/queries/relation_tree/_find_first.py

@@ -38,7 +38,7 @@
 from ._base import RelationBase, StringOrWildcard
 
 if TYPE_CHECKING:
-    from ._relation import Relation
+    from ._select import Select
 
 
 class FindFirst(RelationBase):
@@ -51,7 +51,7 @@ class FindFirst(RelationBase):
 
     relation_type: Literal["find_first"] = "find_first"
 
-    operand: Relation
+    operand: Select
     """The upstream relation to operate on.
 
     This may have more than one `DatasetSearch` joined into it (at any level),

python/lsst/daf/butler/queries/relation_tree/_materialization.py

@@ -37,7 +37,7 @@
 from ._base import RelationBase, StringOrWildcard
 
 if TYPE_CHECKING:
-    from ._relation import Relation
+    from ._relation import RootRelation
 
 
 class Materialization(RelationBase):
@@ -47,7 +47,7 @@ class Materialization(RelationBase):
 
     relation_type: Literal["materialization"] = "materialization"
 
-    operand: Relation
+    operand: RootRelation
     """The upstream relation to evaluate."""
 
     dataset_types: frozenset[StringOrWildcard]

python/lsst/daf/butler/queries/relation_tree/_ordered_slice.py

@@ -44,7 +44,7 @@
 )
 
 if TYPE_CHECKING:
-    from ._relation import Relation
+    from ._relation import OrderedSliceOperand
 
 
 class OrderedSlice(RelationBase):
@@ -54,7 +54,7 @@ class OrderedSlice(RelationBase):
 
     relation_type: Literal["ordered_slice"] = "ordered_slice"
 
-    operand: Relation
+    operand: OrderedSliceOperand
     """The upstream relation to operate on."""
 
     order_by: tuple[OrderExpression, ...] = ()

python/lsst/daf/butler/queries/relation_tree/_predicate.py

@@ -40,7 +40,7 @@
 from ._column_expression import ColumnExpression
 
 if TYPE_CHECKING:
-    from ._relation import Relation
+    from ._relation import RootRelation
 
 
 class LogicalAnd(PredicateBase):
@@ -148,7 +148,7 @@ class InRelation(PredicateBase):
     predicate_type: Literal["in_relation"] = "in_relation"
     member: ColumnExpression
     column: ColumnExpression
-    relation: Relation
+    relation: RootRelation
 
     def gather_required_columns(self) -> set[ColumnReference]:
         # We're only gathering columns from the relation this predicate is

python/lsst/daf/butler/queries/relation_tree/_relation.py

@@ -27,7 +27,7 @@
 
 from __future__ import annotations
 
-__all__ = ("Relation", "DeferredValidationRelation")
+__all__ = ("RootRelation", "DeferredValidationRootRelation", "JoinOperand", "OrderedSliceOperand")
 
 from typing import Annotated, TypeAlias, Union
 
@@ -36,32 +36,37 @@
 from ...pydantic_utils import DeferredValidation
 from ._data_coordinate_upload import DataCoordinateUpload
 from ._dataset_search import DatasetSearch
-from ._dimension_join import DimensionJoin
 from ._find_first import FindFirst
 from ._materialization import Materialization
 from ._ordered_slice import OrderedSlice
-from ._selection import Selection
+from ._select import Select
 
-Relation: TypeAlias = Annotated[
+JoinOperand: TypeAlias = Annotated[
     Union[
         DataCoordinateUpload,
         DatasetSearch,
-        DimensionJoin,
-        FindFirst,
         Materialization,
-        OrderedSlice,
-        Selection,
     ],
     pydantic.Field(discriminator="relation_type"),
 ]
 
+OrderedSliceOperand: TypeAlias = Annotated[
+    Union[Select, FindFirst],
+    pydantic.Field(discriminator="relation_type"),
+]
+
+
+RootRelation: TypeAlias = Annotated[
+    Union[Select, FindFirst, OrderedSlice],
+    pydantic.Field(discriminator="relation_type"),
+]
+
 
-DimensionJoin.model_rebuild()
+Select.model_rebuild()
 FindFirst.model_rebuild()
 Materialization.model_rebuild()
 OrderedSlice.model_rebuild()
-Selection.model_rebuild()
 
 
-class DeferredValidationRelation(DeferredValidation[Relation]):
+class DeferredValidationRootRelation(DeferredValidation[RootRelation]):
     pass

python/lsst/daf/butler/queries/relation_tree/_dimension_join.py → python/lsst/daf/butler/queries/relation_tree/_select.py

@@ -27,7 +27,7 @@
 
 from __future__ import annotations
 
-__all__ = ("DimensionJoin", "make_unit_relation")
+__all__ = ("Select", "make_unit_relation")
 
 import itertools
 from functools import cached_property
@@ -37,42 +37,44 @@
 
 from ...dimensions import DimensionGroup, DimensionUniverse
 from ._base import RelationBase, StringOrWildcard
+from ._column_reference import DatasetFieldReference, DimensionFieldReference, DimensionKeyReference
+from ._predicate import Predicate
 from .joins import JoinTuple
 
 if TYPE_CHECKING:
-    from ._relation import Relation
+    from ._relation import JoinOperand
 
 
-def make_unit_relation(universe: DimensionUniverse) -> DimensionJoin:
+def make_unit_relation(universe: DimensionUniverse) -> Select:
     """Make an initial relation with empty dimensions and a single logical row.
 
     This method should be used by `Butler._query` to construct the initial
     relation tree.  This relation is a useful initial state because it is the
     identity relation for joins, in that joining any other relation to this
     relation yields that relation.
     """
-    return DimensionJoin.model_construct(dimensions=universe.empty.as_group())
+    return Select.model_construct(dimensions=universe.empty.as_group())
 
 
-class DimensionJoin(RelationBase):
-    """An abstract relation that represents a join between dimension-element
-    tables and (optionally) other relations.
+class Select(RelationBase):
+    """An abstract relation that combines joins and row-selection within a
+    fixed set of dimensions.
 
     Notes
     -----
     Joins on dataset IDs are expected to be expressed as
-    `abstract_expressions.InRelation` predicates in `Selection` operations.
+    `abstract_expressions.InRelation` predicates used in the `where` attribute.
     That is slightly more powerful (since it can do set differences via
     `abstract_expressions.LogicalNot`) and it keeps the abstract relation tree
-    simpler if the only join constraints in play are on dimension columns.
+    simpler if the only join constraints in play here are on dimension columns.
     """
 
-    relation_type: Literal["dimension_join"] = "dimension_join"
+    relation_type: Literal["select"] = "select"
 
     dimensions: DimensionGroup
     """The dimensions of the relation."""
 
-    operands: tuple[Relation, ...] = ()
+    join_operands: tuple[JoinOperand, ...] = ()
     """Relations to include in the join other than dimension-element tables.
 
     Because dimension-element tables are expected to contain the full set of
@@ -85,29 +87,32 @@ class DimensionJoin(RelationBase):
     ``detector`` table, but the ``physical_filter`` table will be joined in.
     """
 
-    spatial: frozenset[JoinTuple] = frozenset()
+    spatial_joins: frozenset[JoinTuple] = frozenset()
     """Pairs of dimension element names that should whose regions on the sky
     must overlap.
     """
 
-    temporal: frozenset[JoinTuple] = frozenset()
+    temporal_joins: frozenset[JoinTuple] = frozenset()
     """Pairs of dimension element names and calibration dataset type names
     whose timespans must overlap.
     """
 
+    where: tuple[Predicate, ...] = ()
+    """Boolean expression trees whose logical AND defines a row filter."""
+
     @cached_property
     def available_dataset_types(self) -> frozenset[StringOrWildcard]:
         """The dataset types whose ID columns (at least) are available from
         this relation.
         """
         result: set[StringOrWildcard] = set()
-        for operand in self.operands:
+        for operand in self.join_operands:
             result.update(operand.available_dataset_types)
         return frozenset(result)
 
     @pydantic.model_validator(mode="after")
-    def _validate_operands(self) -> DimensionJoin:
-        for operand in self.operands:
+    def _validate_join_operands(self) -> Select:
+        for operand in self.join_operands:
             if not operand.dimensions.issubset(self.dimensions):
                 raise ValueError(
                     f"Dimensions {operand.dimensions} of join operand {operand} are not a "
@@ -116,7 +121,7 @@ def _validate_operands(self) -> DimensionJoin:
         return self
 
     @pydantic.model_validator(mode="after")
-    def _validate_spatial(self) -> DimensionJoin:
+    def _validate_spatial_joins(self) -> Select:
         def check_operand(operand: str) -> str:
             if operand not in self.dimensions.elements:
                 raise ValueError(f"Spatial join operand {operand!r} is not in this join's dimensions.")
@@ -125,13 +130,13 @@ def check_operand(operand: str) -> str:
                 raise ValueError(f"Spatial join operand {operand!r} is not associated with a region.")
             return family.name
 
-        for a, b in self.spatial:
+        for a, b in self.spatial_joins:
             if check_operand(a) == check_operand(b):
                 raise ValueError(f"Spatial join between {a!r} and {b!r} is unnecessary.")
         return self
 
     @pydantic.model_validator(mode="after")
-    def _validate_temporal(self) -> DimensionJoin:
+    def _validate_temporal_joins(self) -> Select:
         def check_operand(operand: str) -> str:
             if operand in self.dimensions.elements:
                 family = self.dimensions.universe[operand].temporal
@@ -145,18 +150,36 @@ def check_operand(operand: str) -> str:
                     f"Temporal join operand {operand!r} is not in this join's dimensions or dataset types."
                 )
 
-        for a, b in self.spatial:
+        for a, b in self.spatial_joins:
             if check_operand(a) == check_operand(b):
                 raise ValueError(f"Temporal join between {a!r} and {b!r} is unnecessary or impossible.")
         return self
 
     @pydantic.model_validator(mode="after")
-    def _validate_upstream_datasets(self) -> DimensionJoin:
-        for a, b in itertools.combinations(self.operands, 2):
+    def _validate_upstream_datasets(self) -> Select:
+        for a, b in itertools.combinations(self.join_operands, 2):
             if not a.available_dataset_types.isdisjoint(b.available_dataset_types):
                 common = a.available_dataset_types & b.available_dataset_types
                 if None in common:
                     raise ValueError(f"Upstream relations {a} and {b} both have a dataset wildcard.")
                 else:
                     raise ValueError(f"Upstream relations {a} and {b} both have dataset types {common}.")
         return self
+
+    @pydantic.model_validator(mode="after")
+    def _validate_required_columns(self) -> Select:
+        required_columns = set()
+        for where_term in self.where:
+            required_columns.update(where_term.gather_required_columns())
+        for column in required_columns:
+            match column:
+                case DimensionKeyReference(dimension=dimension):
+                    if dimension not in self.dimensions:
+                        raise ValueError(f"Column {column} is not in dimensions {self.dimensions}.")
+                case DimensionFieldReference(element=element):
+                    if element not in self.dimensions.elements:
+                        raise ValueError(f"Column {column} is not in dimensions {self.dimensions}.")
+                case DatasetFieldReference(dataset_type=dataset_type):
+                    if dataset_type not in self.available_dataset_types:
+                        raise ValueError(f"Dataset search for column {column} is not present.")
+        return self