Add commutator for FindFirstDataset.

I don't know why I previously assumed this wouldn't commute with
anything; it commutes in essentially exactly the same way as a
Selection (i.e. WHERE clause).

doc/changes/DM-40184.bugfix.md

@@ -0,0 +1,3 @@
+Fix a rare bug in follow-up dataset queries involving relation commutators.
+
+This occurred when building QuantumGraphs where a "warp" dataset type was an overall input to the pipeline and present in more than one input RUN collection.

python/lsst/daf/butler/registry/queries/find_first_dataset.py

@@ -26,7 +26,7 @@
 from collections.abc import Sequence, Set
 from typing import final
 
-from lsst.daf.relation import ColumnTag, Relation, RowFilter
+from lsst.daf.relation import ColumnTag, Relation, RowFilter, UnaryCommutator, UnaryOperationRelation
 from lsst.utils.classes import cached_getter
 
 from ...core import DatasetColumnTag, DimensionKeyColumnTag
@@ -73,3 +73,24 @@ def __str__(self) -> str:
     def applied_min_rows(self, target: Relation) -> int:
         # Docstring inherited.
         return 1 if target.min_rows else 0
+
+    def commute(self, current: UnaryOperationRelation) -> UnaryCommutator:
+        # Docstring inherited.
+        if not self.columns_required <= current.target.columns:
+            return UnaryCommutator(
+                first=None,
+                second=current.operation,
+                done=False,
+                messages=(
+                    f"{current.target} is missing columns "
+                    f"{set(self.columns_required - current.target.columns)}",
+                ),
+            )
+        if current.operation.is_count_dependent:
+            return UnaryCommutator(
+                first=None,
+                second=current.operation,
+                done=False,
+                messages=(f"{current.operation} is count-dependent",),
+            )
+        return UnaryCommutator(self, current.operation)

python/lsst/daf/butler/registry/tests/_registry.py

@@ -3428,7 +3428,6 @@ def pop_transfer(tree: Relation) -> Relation:
             sql.Engine,
         )
 
-    @unittest.expectedFailure
     def test_query_find_datasets_drop_postprocessing(self) -> None:
         """Test that DataCoordinateQueryResults.findDatasets will strip
         postprocessing operations when it needs to.