DM-55542: Ensure that local butler has same universe as source butler

doc/changes/DM-55542.bugfix.md

@@ -0,0 +1,3 @@
+`PipelineGraph.register_dataset_types` now conforms each dataset type to the butler's dimension universe before registration.
+This allows a quantum graph resolved against one universe version to be executed with a butler using a different but compatible universe, such as a local butler repository created by an earlier run.
+`SimplePipelineExecutor.use_local_butler` also now creates the local data repository with the source butler's dimension universe.

python/lsst/pipe/base/pipeline_graph/_pipeline_graph.py

@@ -1740,7 +1740,7 @@ def register_dataset_types(
                 if task_node.log_output is not None:
                     del dataset_types[task_node.log_output.dataset_type_name]
         for dataset_type in dataset_types.values():
-            butler.registry.registerDatasetType(dataset_type)
+            butler.registry.registerDatasetType(dataset_type.conform_to(butler.dimensions))
 
     def check_dataset_type_registrations(self, butler: Butler, include_packages: bool = True) -> None:
         """Check that dataset type registrations in a data repository match

python/lsst/pipe/base/simple_pipeline_executor.py

@@ -507,7 +507,9 @@ def use_local_butler(
         collections.
         """
         if not Butler.has_repo_config(root):
-            Butler.makeRepo(root)
+            # Ensure that the universe in the output butler matches the
+            # universe in the source butler for maximal compatibility.
+            Butler.makeRepo(root, dimensionConfig=self.butler.dimensions.dimensionConfig)
         out_butler = Butler.from_config(root, writeable=True)
 
         output_run = self.predicted.header.output_run

requirements.txt

@@ -7,7 +7,7 @@ wcwidth
 frozendict
 astropy
 zstandard >=0.23.0,<0.24
-lsst-daf-butler @ git+https://github.com/lsst/daf_butler@main
+lsst-daf-butler @ git+https://github.com/lsst/daf_butler@tickets/DM-55542
 lsst-utils @ git+https://github.com/lsst/utils@main
 lsst-resources @ git+https://github.com/lsst/resources@main
 lsst-pex-config @ git+https://github.com/lsst/pex_config@main

tests/test_pipeline_graph.py

@@ -38,7 +38,15 @@
 
 import lsst.pipe.base.automatic_connection_constants as acc
 import lsst.utils.tests
-from lsst.daf.butler import DataCoordinate, DatasetRef, DatasetType, DimensionUniverse, StorageClassFactory
+from lsst.daf.butler import (
+    DataCoordinate,
+    DatasetRef,
+    DatasetType,
+    DimensionConfig,
+    DimensionUniverse,
+    InconsistentUniverseError,
+    StorageClassFactory,
+)
 from lsst.daf.butler.registry import MissingDatasetTypeError
 from lsst.pipe.base.pipeline_graph import (
     ConnectionTypeConsistencyError,
@@ -60,6 +68,7 @@
     DynamicConnectionConfig,
     DynamicTestPipelineTask,
     DynamicTestPipelineTaskConfig,
+    InMemoryRepo,
     get_mock_name,
 )
 
@@ -1830,6 +1839,65 @@ def test_invalid_dataset_type_name(self) -> None:
         self.assertEqual(error.exception.__notes__, ["In connection 'o' of task 'a'."])
 
 
+class RegisterDatasetTypesUniverseTestCase(unittest.TestCase):
+    """Tests for `PipelineGraph.register_dataset_types` when the pipeline
+    graph was resolved against a dimension universe that differs from the
+    target butler's universe.
+    """
+
+    def _make_other_universe(self, version_offset: int) -> DimensionUniverse:
+        """Make a dimension universe that differs from the default universe
+        only in its version.
+        """
+        config = DimensionConfig()
+        config["version"] = config["version"] + version_offset
+        return DimensionUniverse(config)
+
+    def test_compatible_universe(self) -> None:
+        """Test that dataset types resolved in a different but compatible
+        universe are registered using the butler's own universe.
+        """
+        universe = self._make_other_universe(1000000)
+        with InMemoryRepo() as helper:
+            self.assertIsNot(universe, helper.butler.dimensions)
+            helper.add_task(dimensions=["detector"])
+            helper.pipeline_graph.resolve(dimensions=universe)
+            helper.pipeline_graph.register_dataset_types(helper.butler)
+            dataset_type = helper.butler.get_dataset_type("dataset_auto0")
+            self.assertIs(dataset_type.dimensions.universe, helper.butler.dimensions)
+            self.assertEqual(dataset_type.dimensions.names, {"instrument", "detector"})
+
+    def test_incompatible_dimensions(self) -> None:
+        """Test that dataset types whose dimensions do not conform identically
+        in the butler's universe are rejected.
+        """
+        config = DimensionConfig()
+        config["version"] = config["version"] + 1000001
+        # Dropping the implied dependency on band means that a dataset type
+        # with physical_filter dimensions conforms to a larger dimension group
+        # in the default universe than it does in this one.
+        config["elements", "physical_filter", "implies"] = []
+        universe = DimensionUniverse(config)
+        with InMemoryRepo() as helper:
+            helper.add_task(dimensions=["physical_filter"])
+            helper.pipeline_graph.resolve(dimensions=universe)
+            with self.assertRaises(InconsistentUniverseError):
+                helper.pipeline_graph.register_dataset_types(helper.butler)
+
+    def test_incompatible_namespace(self) -> None:
+        """Test that dataset types from a universe with a different namespace
+        are rejected.
+        """
+        config = DimensionConfig()
+        config["namespace"] = "pipe_base_test"
+        universe = DimensionUniverse(config)
+        with InMemoryRepo() as helper:
+            helper.add_task(dimensions=["detector"])
+            helper.pipeline_graph.resolve(dimensions=universe)
+            with self.assertRaises(InconsistentUniverseError):
+                helper.pipeline_graph.register_dataset_types(helper.butler)
+
+
 if __name__ == "__main__":
     lsst.utils.tests.init()
     unittest.main()