From 4df9a17414fe82cb2fef1b750ee502a5f2050fba Mon Sep 17 00:00:00 2001 From: Tim Jenness Date: Fri, 17 Jul 2026 15:34:07 -0700 Subject: [PATCH 1/2] Ensure that local butler has same universe as source butler Mismatches cause problems with dataset type registration and dimension record transfers. --- python/lsst/pipe/base/simple_pipeline_executor.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/python/lsst/pipe/base/simple_pipeline_executor.py b/python/lsst/pipe/base/simple_pipeline_executor.py index e9a9bf9a4..e93d747e6 100644 --- a/python/lsst/pipe/base/simple_pipeline_executor.py +++ b/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 From 5cec056931c6678d01f199e5fcc4b215331c29d1 Mon Sep 17 00:00:00 2001 From: Tim Jenness Date: Fri, 17 Jul 2026 16:59:58 -0700 Subject: [PATCH 2/2] Allow dataset type registration with compatible universes PipelineGraph.register_dataset_types now rebuilds each dataset type in the target butler's dimension universe when the graph was resolved against a different but compatible universe (matching the behavior of Butler.transfer_from). This allows a quantum graph built against one universe version to be executed with a local butler created with another, so long as the dimension groups conform identically. Uses daf_butler conform_to API for dataset type registration Co-Authored-By: Claude Fable 5 --- doc/changes/DM-55542.bugfix.md | 3 + .../base/pipeline_graph/_pipeline_graph.py | 2 +- tests/test_pipeline_graph.py | 70 ++++++++++++++++++- 3 files changed, 73 insertions(+), 2 deletions(-) create mode 100644 doc/changes/DM-55542.bugfix.md diff --git a/doc/changes/DM-55542.bugfix.md b/doc/changes/DM-55542.bugfix.md new file mode 100644 index 000000000..347ed7a95 --- /dev/null +++ b/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. diff --git a/python/lsst/pipe/base/pipeline_graph/_pipeline_graph.py b/python/lsst/pipe/base/pipeline_graph/_pipeline_graph.py index 42f7cd293..1222a8be0 100644 --- a/python/lsst/pipe/base/pipeline_graph/_pipeline_graph.py +++ b/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 diff --git a/tests/test_pipeline_graph.py b/tests/test_pipeline_graph.py index 50975955f..af94df769 100644 --- a/tests/test_pipeline_graph.py +++ b/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()