Coverage for python / lsst / pipe / base / single_quantum_executor.py: 11%
221 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-13 01:15 -0800
1# This file is part of pipe_base.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <http://www.gnu.org/licenses/>.
28__all__ = ["SingleQuantumExecutor"]
30import logging
31import time
32import uuid
33from collections import defaultdict
34from collections.abc import Callable, Mapping
35from itertools import chain
36from typing import Any
38from lsst.daf.butler import (
39 Butler,
40 DatasetRef,
41 DatasetType,
42 LimitedButler,
43 NamedKeyDict,
44 Quantum,
45)
46from lsst.daf.butler.logging import ButlerLogRecords
47from lsst.utils.introspection import get_full_type_name
48from lsst.utils.timer import logInfo
50from ._quantumContext import ExecutionResources, QuantumContext
51from ._status import (
52 AnnotatedPartialOutputsError,
53 ExceptionInfo,
54 InvalidQuantumError,
55 NoWorkFound,
56 QuantumSuccessCaveats,
57)
58from .connections import AdjustQuantumHelper
59from .log_capture import LogCapture, _ExecutionLogRecordsExtra
60from .pipeline_graph import TaskNode
61from .pipelineTask import PipelineTask
62from .quantum_graph_executor import QuantumExecutionResult, QuantumExecutor
63from .quantum_reports import QuantumReport
64from .task import _TASK_FULL_METADATA_TYPE, _TASK_METADATA_TYPE
65from .taskFactory import TaskFactory
67_LOG = logging.getLogger(__name__)
70class SingleQuantumExecutor(QuantumExecutor):
71 """Executor class which runs one Quantum at a time.
73 Parameters
74 ----------
75 butler : `~lsst.daf.butler.Butler` or `None`, optional
76 Data butler, `None` means that a limited butler should be used instead.
77 task_factory : `.TaskFactory`, optional
78 Instance of a task factory. Defaults to a new instance of
79 `lsst.pipe.base.TaskFactory`.
80 skip_existing_in : `str` or `~collections.abc.Iterable` [ `str` ]
81 A collection name or list of collections to search for the existing
82 outputs of quanta, which indicates that those quanta should be skipped.
83 This class only checks for the presence of butler output run in the
84 list of collections. If the output run is present in the list then the
85 quanta whose complete outputs exist in the output run will be skipped.
86 `None` or empty string/sequence disables skipping.
87 clobber_outputs : `bool`, optional
88 If `True`, then outputs from a quantum that exist in output run
89 collection will be removed prior to executing a quantum. If
90 ``skip_existing_in`` contains output run, then only partial outputs
91 from a quantum will be removed. Only used when ``butler`` is not
92 `None`.
93 enable_lsst_debug : `bool`, optional
94 Enable debugging with ``lsstDebug`` facility for a task.
95 limited_butler_factory : `~collections.abc.Callable`, optional
96 A method that creates a `~lsst.daf.butler.LimitedButler` instance for a
97 given Quantum. This parameter must be defined if ``butler`` is `None`.
98 If ``butler`` is not `None` then this parameter is ignored.
99 resources : `.ExecutionResources`, optional
100 The resources available to this quantum when executing.
101 skip_existing : `bool`, optional
102 If `True`, skip quanta whose metadata datasets are already stored.
103 Unlike ``skip_existing_in``, this works with limited butlers as well as
104 full butlers. Always set to `True` if ``skip_existing_in`` matches
105 ``butler.run``.
106 assume_no_existing_outputs : `bool`, optional
107 If `True`, assume preexisting outputs are impossible (e.g. because this
108 is known by higher-level code to be a new ``RUN`` collection), and do
109 not look for them. This causes the ``skip_existing`` and
110 ``clobber_outputs`` options to be ignored, but unlike just setting both
111 of those to `False`, it also avoids all dataset existence checks.
112 raise_on_partial_outputs : `bool`, optional
113 If `True` raise exceptions chained by `.AnnotatedPartialOutputsError`
114 immediately, instead of considering the partial result a success and
115 continuing to run downstream tasks.
116 job_metadata : `~collections.abc.Mapping`
117 Mapping with extra metadata to embed within the quantum metadata under
118 the "job" key. This is intended to correspond to information common
119 to all quanta being executed in a single process, such as the time
120 taken to load the quantum graph in a BPS job.
121 """
123 def __init__(
124 self,
125 *,
126 butler: Butler | None = None,
127 task_factory: TaskFactory | None = None,
128 skip_existing_in: Any = None,
129 clobber_outputs: bool = False,
130 enable_lsst_debug: bool = False,
131 limited_butler_factory: Callable[[Quantum], LimitedButler] | None = None,
132 resources: ExecutionResources | None = None,
133 skip_existing: bool = False,
134 assume_no_existing_outputs: bool = False,
135 raise_on_partial_outputs: bool = True,
136 job_metadata: Mapping[str, int | str | float] | None = None,
137 ):
138 self._butler = butler
139 self._task_factory = task_factory if task_factory is not None else TaskFactory()
140 self._clobber_outputs = clobber_outputs
141 self._enable_lsst_debug = enable_lsst_debug
142 self._limited_butler_factory = limited_butler_factory
143 self._resources = resources
144 self._assume_no_existing_outputs = assume_no_existing_outputs
145 self._raise_on_partial_outputs = raise_on_partial_outputs
146 self._job_metadata = job_metadata
148 if self._butler is None:
149 assert limited_butler_factory is not None, "limited_butler_factory is needed when butler is None"
151 # Find whether output run is in skip_existing_in.
152 self._skip_existing = skip_existing
153 if self._butler is not None and skip_existing_in and not self._skip_existing:
154 self._skip_existing = self._butler.run in self._butler.collections.query(
155 skip_existing_in, flatten_chains=True
156 )
157 self._previous_process_quanta: list[uuid.UUID] = []
159 def execute(
160 self,
161 task_node: TaskNode,
162 /,
163 quantum: Quantum,
164 quantum_id: uuid.UUID | None = None,
165 *,
166 log_records: ButlerLogRecords | None = None,
167 ) -> QuantumExecutionResult:
168 # Docstring inherited from QuantumExecutor.execute
169 if self._butler is not None:
170 self._butler.registry.refresh()
171 return self._execute(task_node, quantum, quantum_id=quantum_id, log_records=log_records)
173 def _execute(
174 self,
175 task_node: TaskNode,
176 /,
177 quantum: Quantum,
178 quantum_id: uuid.UUID | None = None,
179 *,
180 log_records: ButlerLogRecords | None = None,
181 ) -> QuantumExecutionResult:
182 """Execute the quantum.
184 Internal implementation of `execute()`.
185 """
186 # Make a limited butler instance if needed.
187 limited_butler: LimitedButler
188 used_butler_factory = False
189 if self._butler is not None:
190 limited_butler = self._butler
191 else:
192 # We check this in constructor, but mypy needs this check here.
193 assert self._limited_butler_factory is not None
194 limited_butler = self._limited_butler_factory(quantum)
195 used_butler_factory = True
197 try:
198 return self._execute_with_limited_butler(
199 task_node, limited_butler, quantum=quantum, quantum_id=quantum_id, log_records=log_records
200 )
201 finally:
202 if used_butler_factory:
203 limited_butler.close()
205 def _execute_with_limited_butler(
206 self,
207 task_node: TaskNode,
208 limited_butler: LimitedButler,
209 /,
210 quantum: Quantum,
211 quantum_id: uuid.UUID | None = None,
212 *,
213 log_records: ButlerLogRecords | None = None,
214 ) -> QuantumExecutionResult:
215 startTime = time.time()
216 assert quantum.dataId is not None, "Quantum DataId cannot be None"
217 report = QuantumReport(quantumId=quantum_id, dataId=quantum.dataId, taskLabel=task_node.label)
218 if self._butler is not None:
219 log_capture = LogCapture.from_full(self._butler)
220 else:
221 log_capture = LogCapture.from_limited(limited_butler)
222 with log_capture.capture_logging(task_node, quantum, records=log_records) as captureLog:
223 # Save detailed resource usage before task start to metadata.
224 quantumMetadata = _TASK_METADATA_TYPE()
225 logInfo(None, "prep", metadata=quantumMetadata) # type: ignore[arg-type]
227 _LOG.info(
228 "Preparing execution of quantum for label=%s dataId=%s.", task_node.label, quantum.dataId
229 )
231 # check whether to skip or delete old outputs, if it returns True
232 # or raises an exception do not try to store logs, as they may be
233 # already in butler.
234 captureLog.store = False
235 if self._check_existing_outputs(quantum, task_node, limited_butler, captureLog.extra):
236 _LOG.info(
237 "Skipping already-successful quantum for label=%s dataId=%s.",
238 task_node.label,
239 quantum.dataId,
240 )
241 return QuantumExecutionResult(quantum, report, skipped_existing=True, adjusted_no_work=False)
242 captureLog.store = True
244 captureLog.extra.previous_process_quanta.extend(self._previous_process_quanta)
245 if quantum_id is not None:
246 self._previous_process_quanta.append(quantum_id)
247 try:
248 quantum = self._updated_quantum_inputs(quantum, task_node, limited_butler)
249 except NoWorkFound as exc:
250 _LOG.info(
251 "Nothing to do for task '%s' on quantum %s; saving metadata and skipping: %s",
252 task_node.label,
253 quantum.dataId,
254 str(exc),
255 )
256 quantumMetadata["caveats"] = QuantumSuccessCaveats.from_adjust_quantum_no_work().value
257 quantumMetadata["outputs"] = []
258 # Make empty metadata that looks something like what a
259 # do-nothing task would write (but we don't bother with empty
260 # nested PropertySets for subtasks). This is slightly
261 # duplicative with logic in pipe_base that we can't easily call
262 # from here; we'll fix this on DM-29761.
263 logInfo(None, "end", metadata=quantumMetadata) # type: ignore[arg-type]
264 fullMetadata = _TASK_FULL_METADATA_TYPE()
265 fullMetadata[task_node.label] = _TASK_METADATA_TYPE()
266 fullMetadata["quantum"] = quantumMetadata
267 if self._job_metadata is not None:
268 fullMetadata["job"] = self._job_metadata
269 self._write_metadata(quantum, fullMetadata, task_node, limited_butler)
270 return QuantumExecutionResult(
271 quantum,
272 report,
273 skipped_existing=False,
274 adjusted_no_work=True,
275 task_metadata=fullMetadata,
276 )
278 # enable lsstDebug debugging
279 if self._enable_lsst_debug:
280 try:
281 _LOG.debug("Will try to import debug.py")
282 import debug # type: ignore # noqa:F401
283 except ImportError:
284 _LOG.warning("No 'debug' module found.")
286 # Ensure that we are executing a frozen config
287 task_node.config.freeze()
288 logInfo(None, "init", metadata=quantumMetadata) # type: ignore[arg-type]
289 init_input_refs = list(quantum.initInputs.values())
291 _LOG.info(
292 "Constructing task and executing quantum for label=%s dataId=%s.",
293 task_node.label,
294 quantum.dataId,
295 )
296 task = self._task_factory.makeTask(task_node, limited_butler, init_input_refs)
297 logInfo(None, "start", metadata=quantumMetadata) # type: ignore[arg-type]
298 outputs_put: list[uuid.UUID] = []
299 try:
300 with limited_butler.record_metrics() as butler_metrics:
301 caveats = self._run_quantum(
302 task, quantum, task_node, limited_butler, quantum_id=quantum_id, ids_put=outputs_put
303 )
304 except Exception as e:
305 _LOG.error(
306 "Execution of task '%s' on quantum %s failed. Exception %s: %s",
307 task_node.label,
308 quantum.dataId,
309 e.__class__.__name__,
310 str(e),
311 )
312 captureLog.extra.exception = ExceptionInfo(
313 type_name=get_full_type_name(e),
314 message=str(e),
315 metadata={},
316 )
317 raise
318 else:
319 quantumMetadata["butler_metrics"] = butler_metrics.model_dump()
320 quantumMetadata["caveats"] = caveats.value
321 # Stringify the UUID for easier compatibility with
322 # PropertyList.
323 finally:
324 logInfo(None, "end", metadata=quantumMetadata) # type: ignore[arg-type]
325 fullMetadata = task.getFullMetadata()
326 quantumMetadata["outputs"] = [str(output) for output in outputs_put]
327 fullMetadata["quantum"] = quantumMetadata
328 if self._job_metadata is not None:
329 fullMetadata["job"] = self._job_metadata
330 captureLog.extra.metadata = fullMetadata
331 self._write_metadata(quantum, fullMetadata, task_node, limited_butler)
332 stopTime = time.time()
333 _LOG.info(
334 "Execution of task '%s' on quantum %s took %.3f seconds",
335 task_node.label,
336 quantum.dataId,
337 stopTime - startTime,
338 )
339 return QuantumExecutionResult(
340 quantum,
341 report,
342 skipped_existing=False,
343 adjusted_no_work=False,
344 task_metadata=fullMetadata,
345 )
347 def _check_existing_outputs(
348 self,
349 quantum: Quantum,
350 task_node: TaskNode,
351 /,
352 limited_butler: LimitedButler,
353 log_extra: _ExecutionLogRecordsExtra,
354 ) -> bool:
355 """Decide whether this quantum needs to be executed.
357 If only partial outputs exist then they are removed if
358 ``clobberOutputs`` is True, otherwise an exception is raised.
360 The ``LimitedButler`` is used for everything, and should be set to
361 ``self.butler`` if no separate ``LimitedButler`` is available.
363 Parameters
364 ----------
365 quantum : `~lsst.daf.butler.Quantum`
366 Quantum to check for existing outputs.
367 task_node : `~.pipeline_graph.TaskNode`
368 Task definition structure.
369 limited_butler : `~lsst.daf.butler.LimitedButler`
370 Butler to use for querying and clobbering.
371 log_extra : `.log_capture.TaskLogRecordsExtra`
372 Extra information to attach to log records.
374 Returns
375 -------
376 exist : `bool`
377 `True` if ``self.skipExisting`` is defined, and a previous
378 execution of this quanta appears to have completed successfully
379 (either because metadata was written or all datasets were written).
380 `False` otherwise.
382 Raises
383 ------
384 RuntimeError
385 Raised if some outputs exist and some not.
386 """
387 if self._assume_no_existing_outputs:
388 return False
390 if self._skip_existing:
391 _LOG.debug(
392 "Checking existence of metadata from previous execution of label=%s dataId=%s.",
393 task_node.label,
394 quantum.dataId,
395 )
396 # Metadata output exists; this is sufficient to assume the previous
397 # run was successful and should be skipped.
398 [metadata_ref] = quantum.outputs[task_node.metadata_output.dataset_type_name]
399 if metadata_ref is not None:
400 if limited_butler.stored(metadata_ref):
401 return True
403 # Find and prune (partial) outputs if `self.clobberOutputs` is set.
404 _LOG.debug(
405 "Looking for existing outputs in the way for label=%s dataId=%s.", task_node.label, quantum.dataId
406 )
407 ref_dict = limited_butler.stored_many(chain.from_iterable(quantum.outputs.values()))
408 if task_node.log_output is not None:
409 (log_ref,) = quantum.outputs[task_node.log_output.dataset_type_name]
410 if ref_dict[log_ref]:
411 _LOG.debug(
412 "Attaching logs from previous attempt on label=%s dataId=%s.",
413 task_node.label,
414 quantum.dataId,
415 )
416 log_extra.attach_previous_attempt(limited_butler.get(log_ref))
417 existingRefs = [ref for ref, exists in ref_dict.items() if exists]
418 missingRefs = [ref for ref, exists in ref_dict.items() if not exists]
419 if existingRefs:
420 if not missingRefs:
421 # Full outputs exist.
422 if self._skip_existing:
423 return True
424 elif self._clobber_outputs:
425 _LOG.info("Removing complete outputs for quantum %s: %s", quantum, existingRefs)
426 limited_butler.pruneDatasets(existingRefs, disassociate=True, unstore=True, purge=True)
427 else:
428 raise RuntimeError(
429 f"Complete outputs exists for a quantum {quantum} "
430 "and neither clobberOutputs nor skipExisting is set: "
431 f"existingRefs={existingRefs}"
432 )
433 else:
434 # Partial outputs from a failed quantum.
435 _LOG.debug(
436 "Partial outputs exist for quantum %s existingRefs=%s missingRefs=%s",
437 quantum,
438 existingRefs,
439 missingRefs,
440 )
441 if self._clobber_outputs:
442 # only prune
443 _LOG.info("Removing partial outputs for task %s: %s", task_node.label, existingRefs)
444 limited_butler.pruneDatasets(existingRefs, disassociate=True, unstore=True, purge=True)
445 return False
446 else:
447 raise RuntimeError(
448 "Registry inconsistency while checking for existing quantum outputs:"
449 f" quantum={quantum} existingRefs={existingRefs}"
450 f" missingRefs={missingRefs}"
451 )
453 # By default always execute.
454 return False
456 def _updated_quantum_inputs(
457 self, quantum: Quantum, task_node: TaskNode, /, limited_butler: LimitedButler
458 ) -> Quantum:
459 """Update quantum with extra information, returns a new updated
460 Quantum.
462 Some methods may require input DatasetRefs to have non-None
463 ``dataset_id``, but in case of intermediate dataset it may not be
464 filled during QuantumGraph construction. This method will retrieve
465 missing info from registry.
467 Parameters
468 ----------
469 quantum : `~lsst.daf.butler.Quantum`
470 Single Quantum instance.
471 task_node : `~.pipeline_graph.TaskNode`
472 Task definition structure.
473 limited_butler : `~lsst.daf.butler.LimitedButler`
474 Butler to use for querying.
476 Returns
477 -------
478 update : `~lsst.daf.butler.Quantum`
479 Updated Quantum instance.
480 """
481 anyChanges = False
482 updatedInputs: defaultdict[DatasetType, list] = defaultdict(list)
483 for key, refsForDatasetType in quantum.inputs.items():
484 _LOG.debug(
485 "Checking existence of input '%s' for label=%s dataId=%s.",
486 key.name,
487 task_node.label,
488 quantum.dataId,
489 )
490 toCheck = []
491 newRefsForDatasetType = updatedInputs[key]
492 for ref in refsForDatasetType:
493 if self._should_assume_exists(quantum, ref):
494 newRefsForDatasetType.append(ref)
495 else:
496 toCheck.append(ref)
497 if not toCheck:
498 _LOG.debug(
499 "Assuming overall input '%s' is present without checks for label=%s dataId=%s.",
500 key.name,
501 task_node.label,
502 quantum.dataId,
503 )
504 continue
505 stored = limited_butler.stored_many(toCheck)
506 for ref in toCheck:
507 if stored[ref]:
508 newRefsForDatasetType.append(ref)
509 else:
510 # This should only happen if a predicted intermediate was
511 # not actually produced upstream, but
512 # datastore misconfigurations can unfortunately also land
513 # us here.
514 _LOG.info("No dataset artifact found for %s", ref)
515 continue
516 if len(newRefsForDatasetType) != len(refsForDatasetType):
517 anyChanges = True
518 # If we removed any input datasets, let the task check if it has enough
519 # to proceed and/or prune related datasets that it also doesn't
520 # need/produce anymore. It will raise NoWorkFound if it can't run,
521 # which we'll let propagate up. This is exactly what we run during QG
522 # generation, because a task shouldn't care whether an input is missing
523 # because some previous task didn't produce it, or because it just
524 # wasn't there during QG generation.
525 namedUpdatedInputs = NamedKeyDict[DatasetType, list[DatasetRef]](updatedInputs.items())
526 helper = AdjustQuantumHelper(namedUpdatedInputs, quantum.outputs)
527 if anyChanges:
528 _LOG.debug("Running adjustQuantum for label=%s dataId=%s.", task_node.label, quantum.dataId)
529 assert quantum.dataId is not None, "Quantum DataId cannot be None"
530 helper.adjust_in_place(task_node.get_connections(), label=task_node.label, data_id=quantum.dataId)
531 return Quantum(
532 taskName=quantum.taskName,
533 taskClass=quantum.taskClass,
534 dataId=quantum.dataId,
535 initInputs=quantum.initInputs,
536 inputs=helper.inputs,
537 outputs=helper.outputs,
538 )
540 def _run_quantum(
541 self,
542 task: PipelineTask,
543 quantum: Quantum,
544 task_node: TaskNode,
545 /,
546 limited_butler: LimitedButler,
547 quantum_id: uuid.UUID | None,
548 ids_put: list[uuid.UUID],
549 ) -> QuantumSuccessCaveats:
550 """Execute task on a single quantum.
552 Parameters
553 ----------
554 task : `PipelineTask`
555 Task object.
556 quantum : `~lsst.daf.butler.Quantum`
557 Single Quantum instance.
558 task_node : `~.pipeline_graph.TaskNode`
559 Task definition structure.
560 limited_butler : `~lsst.daf.butler.LimitedButler`
561 Butler to use for dataset I/O.
562 quantum_id : `uuid.UUID` or `None`
563 ID of the quantum being executed.
564 ids_put : list[ `uuid.UUID` ]
565 List to be populated with the dataset IDs that were written by this
566 quantum. This is an output parameter in order to allow it to be
567 populated even when an exception is raised.
569 Returns
570 -------
571 flags : `QuantumSuccessCaveats`
572 Flags that describe qualified successes.
573 """
574 flags = QuantumSuccessCaveats.NO_CAVEATS
576 # Create a butler that operates in the context of a quantum
577 butlerQC = QuantumContext(limited_butler, quantum, resources=self._resources, quantum_id=quantum_id)
579 # Get the input and output references for the task
580 inputRefs, outputRefs = task_node.get_connections().buildDatasetRefs(quantum)
582 # Call task runQuantum() method.
583 try:
584 task.runQuantum(butlerQC, inputRefs, outputRefs)
585 except NoWorkFound as err:
586 # Not an error, just an early exit.
587 _LOG.info(
588 "Task '%s' on quantum %s exited early with no work found: %s.",
589 task_node.label,
590 quantum.dataId,
591 str(err),
592 )
593 flags |= err.FLAGS
594 except AnnotatedPartialOutputsError as caught:
595 error: BaseException
596 if caught.__cause__ is None:
597 _LOG.error(
598 "Incorrect use of AnnotatedPartialOutputsError: no chained exception found.",
599 task_node.label,
600 quantum.dataId,
601 )
602 error = caught
603 else:
604 error = caught.__cause__
605 if self._raise_on_partial_outputs:
606 # Note: this is a real edge case that required some
607 # experimentation: without 'from None' below, this raise would
608 # produce a "while one exception was being handled, another was
609 # raised" traceback involving AnnotatedPartialOutputsError.
610 # With the 'from None', we get just the error chained to it, as
611 # desired.
612 raise error from None
613 else:
614 _LOG.error(
615 "Task '%s' on quantum %s exited with partial outputs; "
616 "considering this a qualified success and proceeding.",
617 task_node.label,
618 quantum.dataId,
619 )
620 _LOG.error(error, exc_info=error)
621 flags |= caught.FLAGS
622 finally:
623 ids_put.extend(output[2] for output in butlerQC.outputsPut)
624 if not butlerQC.outputsPut:
625 flags |= QuantumSuccessCaveats.ALL_OUTPUTS_MISSING
626 if not butlerQC.outputsPut == butlerQC.allOutputs:
627 flags |= QuantumSuccessCaveats.ANY_OUTPUTS_MISSING
628 return flags
630 def _write_metadata(
631 self, quantum: Quantum, metadata: Any, task_node: TaskNode, /, limited_butler: LimitedButler
632 ) -> None:
633 # DatasetRef has to be in the Quantum outputs, can lookup by name
634 try:
635 [ref] = quantum.outputs[task_node.metadata_output.dataset_type_name]
636 except LookupError as exc:
637 raise InvalidQuantumError(
638 "Quantum outputs is missing metadata dataset type "
639 f"{task_node.metadata_output.dataset_type_name};"
640 " this could happen due to inconsistent options between QuantumGraph generation"
641 " and execution"
642 ) from exc
643 limited_butler.put(metadata, ref)
645 def _should_assume_exists(self, quantum: Quantum, ref: DatasetRef) -> bool | None:
646 """Report whether the given dataset can be assumed to exist because
647 some previous check reported that it did.
649 If this is `True` for a dataset does not in fact exist anymore, that's
650 an unexpected problem that we want to raise as an exception, and
651 definitely not a case where some predicted output just wasn't produced.
652 We can't always tell the difference, but in this case we can.
654 Parameters
655 ----------
656 quantum : `Quantum`
657 Quantum being processed.
658 ref : `lsst.daf.butler.DatasetRef`
659 Reference to the input dataset.
661 Returns
662 -------
663 exists : `bool` or `None`
664 `True` if this dataset is definitely an overall input, `False` if
665 some other quantum in the graph is expected to produce it, and
666 `None` if the answer could not be determined.
667 """
668 if quantum.datastore_records:
669 for datastore_record_data in quantum.datastore_records.values():
670 if ref.id in datastore_record_data.records:
671 return True
672 return False
673 return None